diff --git a/Cargo.lock b/Cargo.lock index 429141b1..cd83854c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -225,6 +225,28 @@ dependencies = [ "pin-project-lite", ] +[[package]] +name = "async-stream" +version = "0.3.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b5a71a6f37880a80d1d7f19efd781e4b5de42c88f0722cc13bcb6cc2cfe8476" +dependencies = [ + "async-stream-impl", + "futures-core", + "pin-project-lite", +] + +[[package]] +name = "async-stream-impl" +version = "0.3.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c7c24de15d275a1ecfd47a380fb4d5ec9bfe0933f309ed5e705b775596a3574d" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "async-trait" version = "0.1.89" @@ -4798,6 +4820,7 @@ name = "streamkit-server" version = "0.2.0" dependencies = [ "anyhow", + "async-stream", "async-trait", "aws-lc-rs", "axum", diff --git a/apps/skit/Cargo.toml b/apps/skit/Cargo.toml index dd64b420..26cede45 100644 --- a/apps/skit/Cargo.toml +++ b/apps/skit/Cargo.toml @@ -132,6 +132,7 @@ aws-lc-rs = "1" # For MoQ auth path matching (optional, with moq feature) moq-lite = { version = "0.15.2", optional = true } blake2 = "0.10.6" +async-stream = "0.3.6" [features] default = ["script", "compositor", "moq"] diff --git a/apps/skit/src/lib.rs b/apps/skit/src/lib.rs index 8c33abea..d882d539 100644 --- a/apps/skit/src/lib.rs +++ b/apps/skit/src/lib.rs @@ -7,6 +7,7 @@ pub mod auth; pub mod cli; pub mod config; pub mod file_security; +pub mod log_viewer; pub mod logging; pub mod marketplace; pub mod marketplace_installer; diff --git a/apps/skit/src/log_viewer.rs b/apps/skit/src/log_viewer.rs new file mode 100644 index 00000000..e739a393 --- /dev/null +++ b/apps/skit/src/log_viewer.rs @@ -0,0 +1,773 @@ +// SPDX-FileCopyrightText: © 2025 StreamKit Contributors +// +// SPDX-License-Identifier: MPL-2.0 + +use axum::{ + extract::{Query, State}, + http::{HeaderMap, StatusCode}, + response::{ + sse::{Event, KeepAlive}, + IntoResponse, Sse, + }, + Json, +}; +use serde::{Deserialize, Serialize}; +use std::convert::Infallible; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncSeekExt, BufReader}; +use tracing::{debug, warn}; + +use crate::state::AppState; + +/// Maximum number of lines that can be requested in a single page. +const MAX_LINE_LIMIT: usize = 5000; + +/// Default number of lines per page. +const DEFAULT_LINE_LIMIT: usize = 500; + +/// Interval between file polls during live tail (milliseconds). +const TAIL_POLL_INTERVAL_MS: u64 = 500; + +/// Maximum number of lines to buffer per SSE event during live tail. +const TAIL_MAX_LINES_PER_EVENT: usize = 200; + +/// Query parameters for paginated log retrieval with filtering. +#[derive(Deserialize)] +pub struct LogQuery { + /// Byte offset to start reading from. Default: 0 for forward, end-of-file for backward. + offset: Option, + /// Maximum number of lines to return (default: 500, max: 5000). + limit: Option, + /// Reading direction: "forward" (default) or "backward". + direction: Option, + /// Case-insensitive substring filter applied to each line. + filter: Option, + /// Filter by log level: "error", "warn", "info", "debug", "trace". + level: Option, +} + +/// Response for paginated log reading. +#[derive(Serialize)] +pub struct LogResponse { + /// The log lines (after filtering). + lines: Vec, + /// Byte offset for the next page in the current direction. + next_offset: u64, + /// Whether more data exists in the given direction. + has_more: bool, + /// Total log file size in bytes. + file_size: u64, +} + +/// Query parameters for the live-tail SSE stream. +#[derive(Deserialize)] +pub struct LogStreamQuery { + /// Case-insensitive substring filter applied to each line. + filter: Option, + /// Filter by log level: "error", "warn", "info", "debug", "trace". + level: Option, +} + +/// Resolve the log file path from config, canonicalizing relative paths against cwd. +fn resolve_log_path(config_path: &str) -> Result { + let path = Path::new(config_path); + if path.is_absolute() { + Ok(path.to_path_buf()) + } else { + let cwd = std::env::current_dir().map_err(|e| { + warn!("Failed to get cwd for log path resolution: {e}"); + StatusCode::INTERNAL_SERVER_ERROR + })?; + Ok(cwd.join(path)) + } +} + +/// Check if a log line matches the given level filter. +/// +/// Looks for the tracing level token (e.g. " INFO ", " WARN ") in the line. +/// For JSON-formatted logs, looks for `"level":""`. +fn matches_level(line: &str, level: &str) -> bool { + let level_upper = level.to_ascii_uppercase(); + + // Text format: "2024-01-01T00:00:00Z INFO skit::server: ..." + // The level token is typically surrounded by whitespace. + let text_token = format!(" {level_upper} "); + if line.contains(&text_token) { + return true; + } + + // Also match " INFO " (double-space before, common in tracing output) + let text_token_double = format!(" {level_upper} "); + if line.contains(&text_token_double) { + return true; + } + + // JSON format: {"level":"INFO", ...} + let json_token = format!("\"level\":\"{level_upper}\""); + if line.contains(&json_token) { + return true; + } + + // JSON with lowercase + let json_token_lower = format!("\"level\":\"{}\"", level.to_ascii_lowercase()); + line.contains(&json_token_lower) +} + +/// Check if a line passes both filters (level + substring). +fn line_passes_filters(line: &str, level: Option<&str>, filter: Option<&str>) -> bool { + if let Some(lvl) = level { + if !lvl.is_empty() && !matches_level(line, lvl) { + return false; + } + } + if let Some(f) = filter { + if !f.is_empty() && !line.to_ascii_lowercase().contains(&f.to_ascii_lowercase()) { + return false; + } + } + true +} + +/// Handler: reads a page of log lines from the configured log file. +/// +/// RBAC: requires `access_all_sessions` (admin only). +/// +/// # Errors +/// +/// Returns `StatusCode::FORBIDDEN` if the caller lacks admin permissions, +/// `StatusCode::NOT_FOUND` if file logging is disabled or the log file does not exist, +/// or `StatusCode::INTERNAL_SERVER_ERROR` on I/O failures. +pub async fn get_logs_handler( + State(app_state): State>, + headers: HeaderMap, + Query(query): Query, +) -> Result { + // Check permissions — admin only + let perms = crate::role_extractor::get_permissions(&headers, &app_state); + if !perms.access_all_sessions { + return Err(StatusCode::FORBIDDEN); + } + + // Check that file logging is enabled + if !app_state.config.log.file_enable { + return Err(StatusCode::NOT_FOUND); + } + + let log_path = resolve_log_path(&app_state.config.log.file_path)?; + + if !log_path.exists() { + return Err(StatusCode::NOT_FOUND); + } + + let limit = query.limit.unwrap_or(DEFAULT_LINE_LIMIT).min(MAX_LINE_LIMIT); + + let direction = query.direction.as_deref().unwrap_or("forward"); + + let file = tokio::fs::File::open(&log_path).await.map_err(|e| { + warn!("Failed to open log file {}: {e}", log_path.display()); + StatusCode::INTERNAL_SERVER_ERROR + })?; + + let metadata = file.metadata().await.map_err(|e| { + warn!("Failed to read log file metadata: {e}",); + StatusCode::INTERNAL_SERVER_ERROR + })?; + let file_size = metadata.len(); + + let response = if direction == "backward" { + read_backward( + file, + file_size, + query.offset, + limit, + query.level.as_deref(), + query.filter.as_deref(), + ) + .await? + } else { + read_forward( + file, + file_size, + query.offset.unwrap_or(0), + limit, + query.level.as_deref(), + query.filter.as_deref(), + ) + .await? + }; + + debug!( + lines = response.lines.len(), + next_offset = response.next_offset, + has_more = response.has_more, + file_size = response.file_size, + "Log viewer: served page" + ); + + Ok(Json(response)) +} + +/// Read lines forward from the given byte offset. +async fn read_forward( + mut file: tokio::fs::File, + file_size: u64, + offset: u64, + limit: usize, + level: Option<&str>, + filter: Option<&str>, +) -> Result { + let seek_to = offset.min(file_size); + file.seek(std::io::SeekFrom::Start(seek_to)).await.map_err(|e| { + warn!("Failed to seek log file: {e}"); + StatusCode::INTERNAL_SERVER_ERROR + })?; + + let mut reader = BufReader::new(file); + let mut lines = Vec::with_capacity(limit.min(256)); + let mut bytes_read: u64 = 0; + + // If we started mid-file, check whether the offset lands on a line + // boundary. Peek the byte immediately before seek_to: if it's '\n', + // we're at the start of a complete line and should NOT skip. + // Only skip the first partial line when seek_to is in the middle of a line. + if seek_to > 0 { + let inner = reader.get_mut(); + inner.seek(std::io::SeekFrom::Start(seek_to - 1)).await.map_err(|e| { + warn!("Failed to seek for peek: {e}"); + StatusCode::INTERNAL_SERVER_ERROR + })?; + let mut peek = [0u8; 1]; + let at_line_boundary = inner.read_exact(&mut peek).await.is_ok() && peek[0] == b'\n'; + + // Re-seek to the original position (the peek read 1 byte before + that byte) + inner.seek(std::io::SeekFrom::Start(seek_to)).await.map_err(|e| { + warn!("Failed to re-seek after peek: {e}"); + StatusCode::INTERNAL_SERVER_ERROR + })?; + + if !at_line_boundary { + // Skip the partial line fragment + let mut partial = String::new(); + if reader.read_line(&mut partial).await.map_err(|e| { + warn!("Failed to read partial line: {e}"); + StatusCode::INTERNAL_SERVER_ERROR + })? > 0 + { + bytes_read += partial.len() as u64; + } + } + } + + let mut line_buf = String::new(); + while lines.len() < limit { + line_buf.clear(); + match reader.read_line(&mut line_buf).await { + Ok(0) => break, + Ok(n) => { + bytes_read += n as u64; + let line = line_buf.trim_end_matches('\n').to_string(); + if line_passes_filters(&line, level, filter) { + lines.push(line); + } + }, + Err(e) => { + warn!("Error reading log line: {e}"); + break; + }, + } + } + + let next_offset = seek_to + bytes_read; + let has_more = next_offset < file_size; + + Ok(LogResponse { lines, next_offset, has_more, file_size }) +} + +/// Read lines backward from the given byte offset (or end of file). +/// +/// Reads in chunks working backward from the offset, tracking each line's +/// byte position in the file. This ensures correct pagination across +/// multiple requests: `next_offset` always points to the byte position of +/// the oldest returned line, so the next backward page ends exactly before it. +async fn read_backward( + mut file: tokio::fs::File, + file_size: u64, + offset: Option, + limit: usize, + level: Option<&str>, + filter: Option<&str>, +) -> Result { + let end = offset.unwrap_or(file_size).min(file_size); + + if end == 0 { + return Ok(LogResponse { lines: Vec::new(), next_offset: 0, has_more: false, file_size }); + } + + // Each entry: (byte_offset_in_file, line_text). + let mut collected: Vec<(u64, String)> = Vec::with_capacity(limit.min(256)); + let mut current_end = end; + // Tail fragment carried from the start of the previously-read (higher-offset) + // chunk. It is the continuation of the current chunk's last line. + let mut carry = String::new(); + + while collected.len() < limit && current_end > 0 { + let remaining_lines = limit - collected.len(); + let chunk_size: u64 = (remaining_lines as u64 * 8192).max(32768).min(current_end); + let chunk_start = current_end.saturating_sub(chunk_size); + + file.seek(std::io::SeekFrom::Start(chunk_start)).await.map_err(|e| { + warn!("Failed to seek log file: {e}"); + StatusCode::INTERNAL_SERVER_ERROR + })?; + + let read_len = usize::try_from(current_end - chunk_start).unwrap_or(usize::MAX); + let mut buf = vec![0u8; read_len]; + file.read_exact(&mut buf).await.map_err(|e| { + warn!("Failed to read log file chunk: {e}"); + StatusCode::INTERNAL_SERVER_ERROR + })?; + + // Parse lines from the raw buffer, recording each line's file offset. + let mut chunk_lines: Vec<(u64, String)> = Vec::new(); + let mut seg_start: usize = 0; + for (i, &byte) in buf.iter().enumerate() { + if byte == b'\n' { + let text = String::from_utf8_lossy(&buf[seg_start..i]).into_owned(); + chunk_lines.push((chunk_start + seg_start as u64, text)); + seg_start = i + 1; + } + } + if seg_start < buf.len() { + let text = String::from_utf8_lossy(&buf[seg_start..]).into_owned(); + chunk_lines.push((chunk_start + seg_start as u64, text)); + } + + // The last element extends to current_end. If carry is non-empty it is the + // tail continuation from a higher-offset chunk — join to complete the line. + if !carry.is_empty() { + if let Some(last) = chunk_lines.last_mut() { + last.1.push_str(&carry); + } + carry.clear(); + } + + // If chunk_start > 0, the first element may be a partial line whose + // beginning is in an earlier chunk. Check the byte just before chunk_start: + // if it's '\n', the chunk starts at a line boundary and the first segment + // is a complete line. Only carry it if the preceding byte is NOT '\n'. + if chunk_start > 0 && !chunk_lines.is_empty() { + let mut peek = [0u8; 1]; + let peek_is_newline = + if file.seek(std::io::SeekFrom::Start(chunk_start - 1)).await.is_ok() { + file.read_exact(&mut peek).await.is_ok() && peek[0] == b'\n' + } else { + false + }; + + if !peek_is_newline { + carry = chunk_lines.remove(0).1; + } + } + + // Collect filtered lines in reverse (newest first). + for (pos, text) in chunk_lines.into_iter().rev() { + if !text.is_empty() && line_passes_filters(&text, level, filter) { + collected.push((pos, text)); + if collected.len() >= limit { + break; + } + } + } + + current_end = chunk_start; + } + + // If carry is non-empty the loop reached byte 0 and the first file line + // was saved as carry. Include it if we still have capacity. + if !carry.is_empty() && collected.len() < limit && line_passes_filters(&carry, level, filter) { + collected.push((0, carry)); + } + + // collected is newest-first; reverse to chronological order. + collected.reverse(); + + // next_offset = byte position of the oldest returned line so the next + // backward page ends right before it with no gaps or overlaps. + let next_offset = collected.first().map_or(0, |(pos, _)| *pos); + + Ok(LogResponse { + lines: collected.into_iter().map(|(_, text)| text).collect(), + next_offset, + has_more: next_offset > 0, + file_size, + }) +} + +/// SSE endpoint: streams new log lines as they are appended to the log file. +/// +/// RBAC: requires `access_all_sessions` (admin only). +/// +/// # Errors +/// +/// Returns `StatusCode::FORBIDDEN` if the caller lacks admin permissions, +/// `StatusCode::NOT_FOUND` if file logging is disabled or the log file does not exist, +/// or `StatusCode::INTERNAL_SERVER_ERROR` on I/O failures. +pub async fn stream_logs_handler( + State(app_state): State>, + headers: HeaderMap, + Query(query): Query, +) -> Result>>, StatusCode> { + // Check permissions — admin only + let perms = crate::role_extractor::get_permissions(&headers, &app_state); + if !perms.access_all_sessions { + return Err(StatusCode::FORBIDDEN); + } + + if !app_state.config.log.file_enable { + return Err(StatusCode::NOT_FOUND); + } + + let log_path = resolve_log_path(&app_state.config.log.file_path)?; + + if !log_path.exists() { + return Err(StatusCode::NOT_FOUND); + } + + let filter = query.filter; + let level = query.level; + + let stream = async_stream::stream! { + // Open file and seek to end + let Ok(mut file) = tokio::fs::File::open(&log_path).await else { + yield Ok(Event::default().data("[error] Failed to open log file")); + return; + }; + + let Ok(metadata) = file.metadata().await else { + yield Ok(Event::default().data("[error] Failed to read log file metadata")); + return; + }; + + let mut last_size = metadata.len(); + if file.seek(std::io::SeekFrom::End(0)).await.is_err() { + yield Ok(Event::default().data("[error] Failed to seek to end of log file")); + return; + } + + // Buffer for incomplete trailing line fragments between polls. + let mut tail_carry = String::new(); + + loop { + tokio::time::sleep(tokio::time::Duration::from_millis(TAIL_POLL_INTERVAL_MS)).await; + + // Check current file size + let Ok(metadata) = tokio::fs::metadata(&log_path).await else { + continue; + }; + let current_size = metadata.len(); + + if current_size < last_size { + // File was truncated (rotated) — reopen from start + let Ok(new_file) = tokio::fs::File::open(&log_path).await else { + continue; + }; + file = new_file; + last_size = 0; + tail_carry.clear(); + yield Ok(Event::default().event("truncated").data("Log file was rotated")); + continue; + } + + if current_size == last_size { + continue; + } + + // Read new data + let new_bytes = usize::try_from(current_size - last_size).unwrap_or(usize::MAX); + let mut buf = vec![0u8; new_bytes]; + if file.read_exact(&mut buf).await.is_err() { + // Re-seek if read fails + let _ = file.seek(std::io::SeekFrom::Start(current_size)).await; + last_size = current_size; + continue; + } + + last_size = current_size; + + // Prepend any incomplete line fragment from the previous poll. + let raw = String::from_utf8_lossy(&buf); + let text = if tail_carry.is_empty() { + raw.into_owned() + } else { + let mut combined = std::mem::take(&mut tail_carry); + combined.push_str(&raw); + combined + }; + + // Split into segments. The last segment is either empty (data + // ended with '\n') or an incomplete line to carry forward. + let segments: Vec<&str> = text.split('\n').collect(); + let (complete, trailing) = segments.split_at(segments.len().saturating_sub(1)); + if let Some(&last) = trailing.first() { + if !last.is_empty() { + tail_carry = last.to_string(); + } + } + + let new_lines: Vec<&str> = complete + .iter() + .copied() + .filter(|line| { + !line.is_empty() + && line_passes_filters(line, level.as_deref(), filter.as_deref()) + }) + .collect(); + + if new_lines.is_empty() { + continue; + } + + // Send lines in batches to avoid overwhelming the client + for chunk in new_lines.chunks(TAIL_MAX_LINES_PER_EVENT) { + let payload = chunk.join("\n"); + yield Ok(Event::default().data(payload)); + } + } + }; + + Ok(Sse::new(stream).keep_alive(KeepAlive::default())) +} + +#[cfg(test)] +#[allow(clippy::unwrap_used)] +mod tests { + use super::*; + + #[test] + fn test_matches_level_text_format() { + let line = "2024-01-01T00:00:00Z INFO skit::server: Starting server"; + assert!(matches_level(line, "info")); + assert!(matches_level(line, "INFO")); + assert!(!matches_level(line, "warn")); + assert!(!matches_level(line, "error")); + } + + #[test] + fn test_matches_level_json_format() { + let line = r#"{"timestamp":"2024-01-01T00:00:00Z","level":"WARN","message":"test"}"#; + assert!(matches_level(line, "warn")); + assert!(matches_level(line, "WARN")); + assert!(!matches_level(line, "info")); + } + + #[test] + fn test_line_passes_filters_no_filters() { + assert!(line_passes_filters("any line", None, None)); + } + + #[test] + fn test_line_passes_filters_level_only() { + let line = "2024-01-01T00:00:00Z WARN skit::server: Something happened"; + assert!(line_passes_filters(line, Some("warn"), None)); + assert!(!line_passes_filters(line, Some("error"), None)); + } + + #[test] + fn test_line_passes_filters_text_only() { + let line = "2024-01-01T00:00:00Z INFO skit::server: Starting server"; + assert!(line_passes_filters(line, None, Some("starting"))); + assert!(line_passes_filters(line, None, Some("SERVER"))); + assert!(!line_passes_filters(line, None, Some("shutdown"))); + } + + #[test] + fn test_line_passes_filters_both() { + let line = "2024-01-01T00:00:00Z INFO skit::server: Starting server"; + assert!(line_passes_filters(line, Some("info"), Some("starting"))); + assert!(!line_passes_filters(line, Some("warn"), Some("starting"))); + assert!(!line_passes_filters(line, Some("info"), Some("shutdown"))); + } + + #[test] + fn test_line_passes_filters_empty_strings() { + let line = "any line"; + assert!(line_passes_filters(line, Some(""), None)); + assert!(line_passes_filters(line, None, Some(""))); + assert!(line_passes_filters(line, Some(""), Some(""))); + } + + /// Verify that backward reading correctly joins partial lines at chunk + /// boundaries instead of dropping them. + #[tokio::test] + async fn test_read_backward_multi_chunk_preserves_lines() { + use tokio::io::AsyncWriteExt; + + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("test.log"); + + let mut f = tokio::fs::File::create(&path).await.unwrap(); + let mut expected: Vec = Vec::new(); + for i in 0..20 { + let line = format!("2024-01-01T00:00:00Z INFO test: line number {i}"); + f.write_all(line.as_bytes()).await.unwrap(); + f.write_all(b"\n").await.unwrap(); + expected.push(line); + } + f.flush().await.unwrap(); + drop(f); + + let file = tokio::fs::File::open(&path).await.unwrap(); + let file_size = file.metadata().await.unwrap().len(); + + let resp = read_backward(file, file_size, None, 100, None, None).await.unwrap(); + + assert_eq!(resp.lines, expected, "all lines should be returned in order"); + assert!(!resp.has_more); + } + + /// Verify that paginating backward across multiple requests produces every + /// line exactly once with no truncation or gaps. + #[tokio::test] + async fn test_read_backward_paginated_no_gaps() { + use tokio::io::AsyncWriteExt; + + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("test.log"); + + // Write 20 lines of varying lengths so chunk boundaries are likely to + // fall mid-line. + let mut f = tokio::fs::File::create(&path).await.unwrap(); + let mut expected: Vec = Vec::new(); + for i in 0..20 { + let line = format!( + "2024-01-01T00:00:{i:02}Z INFO test: line {i} padding={}", + "x".repeat(i * 10) + ); + f.write_all(line.as_bytes()).await.unwrap(); + f.write_all(b"\n").await.unwrap(); + expected.push(line); + } + f.flush().await.unwrap(); + drop(f); + + let file_size = tokio::fs::metadata(&path).await.unwrap().len(); + + // Read backward in small pages of 5 and collect all results. + let mut all_lines: Vec = Vec::new(); + let mut offset: Option = None; + let mut pages = 0; + + loop { + let file = tokio::fs::File::open(&path).await.unwrap(); + let resp = read_backward(file, file_size, offset, 5, None, None).await.unwrap(); + + // Prepend: each page is older than the previously collected lines. + let mut page = resp.lines; + page.append(&mut all_lines); + all_lines = page; + + pages += 1; + + if !resp.has_more { + break; + } + offset = Some(resp.next_offset); + + assert!(pages <= 10, "too many pages — possible infinite loop"); + } + + assert_eq!(all_lines, expected, "all lines must appear exactly once across pages"); + assert_eq!(pages, 4, "20 lines / 5 per page = 4 pages"); + } + + /// Verify that read_backward does NOT corrupt lines when a chunk boundary + /// aligns exactly with a line boundary (i.e. the byte before chunk_start + /// is '\n'). This exercises the peek-byte logic that distinguishes partial + /// lines from complete lines at chunk boundaries. + #[tokio::test] + async fn test_read_backward_chunk_boundary_on_line_boundary() { + use tokio::io::AsyncWriteExt; + + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("test.log"); + + // Write enough data to force multiple chunks (>32KB). + // Each line is exactly 100 bytes (99 chars + '\n') so we can + // reason about byte positions precisely. + let mut f = tokio::fs::File::create(&path).await.unwrap(); + let mut expected: Vec = Vec::new(); + let line_count = 500; // 500 * 100 = 50KB, exceeds the 32KB min chunk + for i in 0..line_count { + let prefix = format!("2024-01-01T00:00:00Z INFO test: line {i:04} "); + let padding = "X".repeat(99 - prefix.len()); + let line = format!("{prefix}{padding}"); + assert_eq!(line.len(), 99, "each line should be 99 chars"); + f.write_all(line.as_bytes()).await.unwrap(); + f.write_all(b"\n").await.unwrap(); + expected.push(line); + } + f.flush().await.unwrap(); + drop(f); + + let file = tokio::fs::File::open(&path).await.unwrap(); + let file_size = file.metadata().await.unwrap().len(); + assert_eq!(file_size, line_count as u64 * 100); + + // Read all lines backward in one request. + let resp = read_backward(file, file_size, None, line_count + 10, None, None).await.unwrap(); + + assert_eq!(resp.lines.len(), expected.len(), "should return all {line_count} lines"); + + // Verify no line merging happened — each line should match exactly. + for (i, (got, want)) in resp.lines.iter().zip(expected.iter()).enumerate() { + assert_eq!(got, want, "line {i} mismatch — possible chunk boundary corruption"); + } + } + + /// Verify that paginating forward across multiple requests produces every + /// line exactly once with no gaps (no dropped lines at page boundaries). + #[tokio::test] + async fn test_read_forward_paginated_no_gaps() { + use tokio::io::AsyncWriteExt; + + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("test.log"); + + let mut f = tokio::fs::File::create(&path).await.unwrap(); + let mut expected: Vec = Vec::new(); + for i in 0..12 { + let line = format!("2024-01-01T00:00:{i:02}Z INFO test: forward line {i}"); + f.write_all(line.as_bytes()).await.unwrap(); + f.write_all(b"\n").await.unwrap(); + expected.push(line); + } + f.flush().await.unwrap(); + drop(f); + + let file_size = tokio::fs::metadata(&path).await.unwrap().len(); + + // Read forward in small pages of 3 and collect all results. + let mut all_lines: Vec = Vec::new(); + let mut offset: u64 = 0; + let mut pages = 0; + + loop { + let file = tokio::fs::File::open(&path).await.unwrap(); + let resp = read_forward(file, file_size, offset, 3, None, None).await.unwrap(); + + all_lines.extend(resp.lines); + pages += 1; + + if !resp.has_more { + break; + } + offset = resp.next_offset; + + assert!(pages <= 10, "too many pages — possible infinite loop"); + } + + assert_eq!(all_lines, expected, "all lines must appear exactly once across forward pages"); + assert_eq!(pages, 4, "12 lines / 3 per page = 4 pages"); + } +} diff --git a/apps/skit/src/main.rs b/apps/skit/src/main.rs index 451b76e1..c86df8f7 100644 --- a/apps/skit/src/main.rs +++ b/apps/skit/src/main.rs @@ -35,6 +35,7 @@ mod auth; mod cli; mod config; mod file_security; +mod log_viewer; mod logging; mod marketplace; mod marketplace_installer; diff --git a/apps/skit/src/server.rs b/apps/skit/src/server.rs index ba151091..fe1fce62 100644 --- a/apps/skit/src/server.rs +++ b/apps/skit/src/server.rs @@ -3115,6 +3115,8 @@ pub fn create_app( .route("/api/v1/config", get(get_config_handler)) .route("/api/v1/schema/nodes", get(list_node_definitions_handler)) .route("/api/v1/schema/packets", get(list_packet_types_handler)) + .route("/api/v1/logs", get(crate::log_viewer::get_logs_handler)) + .route("/api/v1/logs/stream", get(crate::log_viewer::stream_logs_handler)) .route("/api/v1/sessions", get(list_sessions_handler).post(create_session_handler)) .route("/api/v1/sessions/{id}", delete(destroy_session_handler)) .route("/api/v1/sessions/{id}/pipeline", get(get_pipeline_handler)) diff --git a/e2e/tests/auth-helpers.ts b/e2e/tests/auth-helpers.ts index 7f85a148..845473a9 100644 --- a/e2e/tests/auth-helpers.ts +++ b/e2e/tests/auth-helpers.ts @@ -66,6 +66,7 @@ export async function ensureLoggedIn(page: Page): Promise { page.getByTestId('convert-view'), page.getByTestId('stream-view'), page.getByTestId('tokens-view'), + page.getByTestId('logs-view'), ]; // Wait for the app to settle on either: diff --git a/e2e/tests/logs.spec.ts b/e2e/tests/logs.spec.ts new file mode 100644 index 00000000..ef6f33dc --- /dev/null +++ b/e2e/tests/logs.spec.ts @@ -0,0 +1,227 @@ +// SPDX-FileCopyrightText: © 2025 StreamKit Contributors +// +// SPDX-License-Identifier: MPL-2.0 + +import { test, expect } from '@playwright/test'; + +import { ensureLoggedIn } from './auth-helpers'; + +test.describe('Log Viewer', () => { + test.beforeEach(async ({ page }) => { + await page.goto('/admin/logs'); + await ensureLoggedIn(page); + if (!page.url().includes('/admin/logs')) { + await page.goto('/admin/logs'); + } + await expect(page.getByTestId('logs-view')).toBeVisible(); + }); + + test('navigates to log viewer and displays UI', async ({ page }) => { + await expect(page.getByTestId('logs-view')).toBeVisible(); + + // Should show the title + await expect(page.getByRole('heading', { name: 'Logs' })).toBeVisible(); + + // Should show the admin nav with Logs link active + await expect(page.getByRole('link', { name: 'Logs' })).toBeVisible(); + + // Log container should be present + await expect(page.getByTestId('logs-container')).toBeVisible(); + }); + + test('filter controls are present and functional', async ({ page }) => { + await expect(page.getByTestId('logs-view')).toBeVisible(); + + // Filter input should be visible + const filterInput = page.getByTestId('logs-filter-input'); + await expect(filterInput).toBeVisible(); + + // Level select should be visible + const levelSelect = page.getByTestId('logs-level-select'); + await expect(levelSelect).toBeVisible(); + + // Page size select should be visible + const pageSizeSelect = page.getByTestId('logs-page-size'); + await expect(pageSizeSelect).toBeVisible(); + + // Wrap toggle should be visible + const wrapToggle = page.getByTestId('logs-wrap-toggle'); + await expect(wrapToggle).toBeVisible(); + + // Expand toggle should be visible + const expandToggle = page.getByTestId('logs-expand-toggle'); + await expect(expandToggle).toBeVisible(); + await expect(expandToggle).toHaveText('Expand'); + + // Live tail button should be visible + const liveTailButton = page.getByTestId('logs-live-tail'); + await expect(liveTailButton).toBeVisible(); + }); + + test('expand toggle switches between constrained and full-width layout', async ({ page }) => { + await expect(page.getByTestId('logs-view')).toBeVisible(); + + const expandToggle = page.getByTestId('logs-expand-toggle'); + + // Initially should show "Expand" (constrained width) + await expect(expandToggle).toHaveText('Expand'); + + // Click to expand + await expandToggle.click(); + await expect(expandToggle).toHaveText('Collapse'); + + // Click again to collapse + await expandToggle.click(); + await expect(expandToggle).toHaveText('Expand'); + }); + + test('clicking a log line copies it to clipboard', async ({ page, context }) => { + await expect(page.getByTestId('logs-view')).toBeVisible(); + + // Grant clipboard permissions + await context.grantPermissions(['clipboard-read', 'clipboard-write']); + + // Wait for log lines to load + const container = page.getByTestId('logs-container'); + await expect(container).toBeVisible(); + + // Get the first log line and click it + const firstLogLine = container.locator('div[title="Click to copy"]').first(); + const lineCount = await container.locator('div[title="Click to copy"]').count(); + + if (lineCount > 0) { + const lineText = await firstLogLine.textContent(); + await firstLogLine.click(); + + // Verify the clipboard contents match + const clipboardText = await page.evaluate(() => navigator.clipboard.readText()); + expect(clipboardText).toBe(lineText); + } + }); + + test('pagination buttons are present', async ({ page }) => { + await expect(page.getByTestId('logs-view')).toBeVisible(); + + // Pagination buttons should be visible + await expect(page.getByTestId('logs-load-older')).toBeVisible(); + await expect(page.getByTestId('logs-load-newer')).toBeVisible(); + await expect(page.getByTestId('logs-load-latest')).toBeVisible(); + }); + + test('level filter applies immediately on change', async ({ page }) => { + await expect(page.getByTestId('logs-view')).toBeVisible(); + + // Select "Error" level filter — should reload without needing a button click + await page.getByTestId('logs-level-select').selectOption('error'); + + // Wait briefly for the filtered results + await page.waitForTimeout(1000); + + // The log container should still be present (may be empty if no errors) + await expect(page.getByTestId('logs-container')).toBeVisible(); + }); + + test('text filter updates as user types', async ({ page }) => { + await expect(page.getByTestId('logs-view')).toBeVisible(); + + // Type a filter — should apply after debounce (no button click needed) + await page.getByTestId('logs-filter-input').fill('skit'); + + // Wait for debounce + request + await page.waitForTimeout(1000); + + // Container should still be visible + await expect(page.getByTestId('logs-container')).toBeVisible(); + }); + + test('log API returns valid response when file logging is enabled', async ({ page }) => { + // Test the logs API directly (cookies from ensureLoggedIn provide auth) + const response = await page.request.get('/api/v1/logs?limit=10&direction=backward'); + + // File logging may be disabled in CI — skip gracefully + if (response.status() === 404) { + test.skip(true, 'File logging is disabled on this server'); + return; + } + + expect(response.ok()).toBeTruthy(); + + const body = (await response.json()) as { + lines: string[]; + next_offset: number; + has_more: boolean; + file_size: number; + }; + + expect(body).toHaveProperty('lines'); + expect(body).toHaveProperty('next_offset'); + expect(body).toHaveProperty('has_more'); + expect(body).toHaveProperty('file_size'); + expect(Array.isArray(body.lines)).toBeTruthy(); + expect(typeof body.next_offset).toBe('number'); + expect(typeof body.has_more).toBe('boolean'); + expect(typeof body.file_size).toBe('number'); + }); + + test('log API supports forward pagination when file logging is enabled', async ({ page }) => { + const firstPage = await page.request.get('/api/v1/logs?limit=5&direction=forward&offset=0'); + + if (firstPage.status() === 404) { + test.skip(true, 'File logging is disabled on this server'); + return; + } + + expect(firstPage.ok()).toBeTruthy(); + + const firstBody = (await firstPage.json()) as { + lines: string[]; + next_offset: number; + has_more: boolean; + }; + + if (firstBody.has_more) { + const secondPage = await page.request.get( + `/api/v1/logs?limit=5&direction=forward&offset=${firstBody.next_offset}` + ); + expect(secondPage.ok()).toBeTruthy(); + + const secondBody = (await secondPage.json()) as { + lines: string[]; + next_offset: number; + }; + + // next_offset should advance + expect(secondBody.next_offset).toBeGreaterThanOrEqual(firstBody.next_offset); + } + }); + + test('log API supports level filtering when file logging is enabled', async ({ page }) => { + const response = await page.request.get('/api/v1/logs?limit=100&direction=backward&level=info'); + + if (response.status() === 404) { + test.skip(true, 'File logging is disabled on this server'); + return; + } + + expect(response.ok()).toBeTruthy(); + + const body = (await response.json()) as { lines: string[] }; + + // All returned lines should contain INFO level marker + for (const line of body.lines) { + const hasInfoLevel = / INFO /i.test(line) || /"level":"INFO"/i.test(line); + expect(hasInfoLevel).toBeTruthy(); + } + }); + + test('admin nav shows Logs link on admin pages', async ({ page }) => { + // Already on /admin/logs from beforeEach + await expect(page.getByTestId('logs-view')).toBeVisible(); + await expect(page.getByRole('link', { name: 'Logs' })).toBeVisible(); + + // Navigate to tokens page (already in ensureLoggedIn's appViews) + await page.goto('/admin/tokens'); + await expect(page.getByTestId('tokens-view')).toBeVisible(); + await expect(page.getByRole('link', { name: 'Logs' })).toBeVisible(); + }); +}); diff --git a/ui/src/App.tsx b/ui/src/App.tsx index ba59e3d3..641c12dc 100644 --- a/ui/src/App.tsx +++ b/ui/src/App.tsx @@ -20,6 +20,7 @@ import { getLogger } from './utils/logger'; import ConvertView from './views/ConvertView'; import DesignView from './views/DesignView'; import LoginView from './views/LoginView'; +import LogsView from './views/LogsView'; import MonitorView from './views/MonitorView'; import PluginsView from './views/PluginsView'; import StreamView from './views/StreamView'; @@ -124,6 +125,7 @@ const App: React.FC = () => { /> } /> } /> + } /> diff --git a/ui/src/services/logs.ts b/ui/src/services/logs.ts new file mode 100644 index 00000000..e2cdb89b --- /dev/null +++ b/ui/src/services/logs.ts @@ -0,0 +1,62 @@ +// SPDX-FileCopyrightText: © 2025 StreamKit Contributors +// +// SPDX-License-Identifier: MPL-2.0 + +import { fetchApi, getApiUrl } from './base'; + +export interface LogResponse { + lines: string[]; + next_offset: number; + has_more: boolean; + file_size: number; +} + +export interface LogQueryParams { + offset?: number; + limit?: number; + direction?: 'forward' | 'backward'; + filter?: string; + level?: string; +} + +/** + * Fetch a page of log lines from the server. + */ +export async function fetchLogs(params: LogQueryParams = {}): Promise { + const searchParams = new URLSearchParams(); + if (params.offset !== undefined) searchParams.set('offset', String(params.offset)); + if (params.limit !== undefined) searchParams.set('limit', String(params.limit)); + if (params.direction) searchParams.set('direction', params.direction); + if (params.filter) searchParams.set('filter', params.filter); + if (params.level) searchParams.set('level', params.level); + + const qs = searchParams.toString(); + const path = qs ? `/api/v1/logs?${qs}` : '/api/v1/logs'; + + const response = await fetchApi(path); + + if (!response.ok) { + if (response.status === 404) { + throw new Error('Log file not available. File logging may be disabled.'); + } + throw new Error(`Failed to fetch logs: ${response.statusText}`); + } + + return response.json(); +} + +/** + * Create an EventSource for live-tailing the log file via SSE. + */ +export function createLogStream(params?: { filter?: string; level?: string }): EventSource { + const searchParams = new URLSearchParams(); + if (params?.filter) searchParams.set('filter', params.filter); + if (params?.level) searchParams.set('level', params.level); + + const qs = searchParams.toString(); + const apiUrl = getApiUrl(); + const base = `${apiUrl}/api/v1/logs/stream`; + const url = qs ? `${base}?${qs}` : base; + + return new EventSource(url, { withCredentials: true }); +} diff --git a/ui/src/views/LogsView.styles.ts b/ui/src/views/LogsView.styles.ts new file mode 100644 index 00000000..06fd49b5 --- /dev/null +++ b/ui/src/views/LogsView.styles.ts @@ -0,0 +1,248 @@ +// SPDX-FileCopyrightText: © 2025 StreamKit Contributors +// +// SPDX-License-Identifier: MPL-2.0 + +import styled from '@emotion/styled'; + +export const Container = styled.div` + box-sizing: border-box; + width: 100%; + min-width: 0; + display: flex; + flex-direction: column; + height: 100%; + min-height: 0; + background: var(--sk-bg); +`; + +export const ContentArea = styled.div` + flex: 1; + display: flex; + justify-content: center; + min-width: 0; + min-height: 0; + overflow: hidden; +`; + +export const ContentWrapper = styled.div<{ $expanded?: boolean }>` + width: 100%; + max-width: ${(props) => (props.$expanded ? 'none' : '1200px')}; + padding: ${(props) => (props.$expanded ? '24px 32px' : '40px')}; + box-sizing: border-box; + display: flex; + flex-direction: column; + min-height: 0; + transition: max-width 0.2s ease; + + @media (max-width: 768px) { + padding: 16px; + } +`; + +export const Card = styled.div` + box-sizing: border-box; + width: 100%; + background: var(--sk-panel-bg); + border: 1px solid var(--sk-border); + border-radius: 12px; + padding: 20px; + display: flex; + flex-direction: column; + gap: 12px; + min-width: 0; + flex: 1; + min-height: 0; +`; + +export const TitleRow = styled.div` + display: flex; + align-items: center; + justify-content: space-between; + gap: 12px; + flex-wrap: wrap; +`; + +export const Title = styled.h1` + margin: 0; + font-size: 20px; + font-weight: 700; + color: var(--sk-text); +`; + +export const Subtle = styled.div` + color: var(--sk-text-muted); + font-size: 13px; +`; + +export const FilterBar = styled.div` + display: flex; + gap: 8px; + align-items: center; + flex-wrap: wrap; +`; + +export const SearchInput = styled.input` + flex: 1; + min-width: 200px; + padding: 8px 12px; + border-radius: 8px; + border: 1px solid var(--sk-border); + background: var(--sk-bg); + color: var(--sk-text); + font-size: 13px; + + &:focus { + outline: none; + border-color: var(--sk-primary); + box-shadow: var(--sk-focus-ring); + } + + &::placeholder { + color: var(--sk-text-muted); + } +`; + +export const LevelSelect = styled.select` + padding: 8px 12px; + border-radius: 8px; + border: 1px solid var(--sk-border); + background: var(--sk-bg); + color: var(--sk-text); + font-size: 13px; +`; + +export const PageSizeSelect = styled.select` + padding: 8px 12px; + border-radius: 8px; + border: 1px solid var(--sk-border); + background: var(--sk-bg); + color: var(--sk-text); + font-size: 13px; +`; + +export const LogContainer = styled.div<{ $wrap?: boolean }>` + flex: 1; + min-height: 120px; + overflow-y: auto; + overflow-x: ${(props) => (props.$wrap !== false ? 'hidden' : 'auto')}; + background: var(--sk-bg); + border: 1px solid var(--sk-border); + border-radius: 8px; + font-family: + ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, 'Liberation Mono', 'Courier New', + monospace; + font-size: 12px; + line-height: 1.5; + padding: 8px 0; + --log-wrap: ${(props) => (props.$wrap !== false ? 'pre-wrap' : 'pre')}; + --log-word-break: ${(props) => (props.$wrap !== false ? 'break-all' : 'normal')}; +`; + +export const LogLine = styled.div<{ $level?: string }>` + white-space: var(--log-wrap, pre-wrap); + word-break: var(--log-word-break, break-all); + padding: 1px 12px; + cursor: pointer; + position: relative; + + &:hover { + background: color-mix(in srgb, var(--sk-primary) 6%, transparent) !important; + } + + ${(props) => { + switch (props.$level) { + case 'error': + return ` + color: var(--sk-danger); + background: color-mix(in srgb, var(--sk-danger) 10%, transparent); + `; + case 'warn': + return ` + color: var(--sk-warning); + background: color-mix(in srgb, var(--sk-warning) 8%, transparent); + `; + case 'debug': + case 'trace': + return 'color: var(--sk-text-muted);'; + default: + return 'color: var(--sk-text);'; + } + }} +`; + +export const PaginationRow = styled.div` + display: flex; + gap: 10px; + align-items: center; + justify-content: space-between; + flex-wrap: wrap; +`; + +export const PaginationInfo = styled.span` + font-size: 12px; + color: var(--sk-text-muted); +`; + +export const EmptyState = styled.div` + display: flex; + align-items: center; + justify-content: center; + padding: 40px; + color: var(--sk-text-muted); + font-size: 14px; +`; + +export const ErrorBox = styled.div` + padding: 12px; + border-radius: 10px; + border: 1px solid var(--sk-border); + background: color-mix(in srgb, var(--sk-danger) 10%, transparent); + color: var(--sk-text); + font-size: 13px; +`; + +export const CopyToast = styled.div<{ $visible: boolean }>` + position: fixed; + bottom: 24px; + right: 24px; + padding: 8px 16px; + border-radius: 8px; + background: var(--sk-panel-bg); + border: 1px solid var(--sk-border); + color: var(--sk-text); + font-size: 13px; + font-weight: 500; + opacity: ${(props) => (props.$visible ? 1 : 0)}; + transform: translateY(${(props) => (props.$visible ? '0' : '8px')}); + transition: + opacity 0.2s ease, + transform 0.2s ease; + pointer-events: none; + z-index: 1000; +`; + +export const LiveIndicator = styled.span<{ $active: boolean }>` + display: inline-flex; + align-items: center; + gap: 6px; + font-size: 12px; + font-weight: 600; + color: ${(props) => (props.$active ? 'var(--sk-success)' : 'var(--sk-text-muted)')}; + + &::before { + content: ''; + display: inline-block; + width: 8px; + height: 8px; + border-radius: 50%; + background: ${(props) => (props.$active ? 'var(--sk-success)' : 'var(--sk-text-muted)')}; + ${(props) => + props.$active + ? `animation: pulse 1.5s ease-in-out infinite; + @keyframes pulse { + 0%, 100% { opacity: 1; } + 50% { opacity: 0.4; } + }` + : ''} + } +`; diff --git a/ui/src/views/LogsView.tsx b/ui/src/views/LogsView.tsx new file mode 100644 index 00000000..ddab4978 --- /dev/null +++ b/ui/src/views/LogsView.tsx @@ -0,0 +1,437 @@ +// SPDX-FileCopyrightText: © 2025 StreamKit Contributors +// +// SPDX-License-Identifier: MPL-2.0 + +import React, { useCallback, useEffect, useRef, useState } from 'react'; +import { flushSync } from 'react-dom'; + +import { Button } from '@/components/ui/Button'; +import { usePermissions } from '@/hooks/usePermissions'; +import { createLogStream, fetchLogs, type LogResponse } from '@/services/logs'; +import { getLogger } from '@/utils/logger'; + +import AdminNav from './admin/AdminNav'; +import { + Card, + Container, + ContentArea, + ContentWrapper, + CopyToast, + EmptyState, + ErrorBox, + FilterBar, + LevelSelect, + LiveIndicator, + LogContainer, + LogLine, + PageSizeSelect, + PaginationInfo, + PaginationRow, + SearchInput, + Subtle, + Title, + TitleRow, +} from './LogsView.styles'; + +const logger = getLogger('LogsView'); + +const DEFAULT_PAGE_SIZE = 500; + +function detectLevel(line: string): string | undefined { + if (/ ERROR /i.test(line) || /"level":"ERROR"/i.test(line)) return 'error'; + if (/ WARN /i.test(line) || /"level":"WARN"/i.test(line)) return 'warn'; + if (/ DEBUG /i.test(line) || /"level":"DEBUG"/i.test(line)) return 'debug'; + if (/ TRACE /i.test(line) || /"level":"TRACE"/i.test(line)) return 'trace'; + return undefined; +} + +function formatFileSize(bytes: number): string { + if (bytes < 1024) return `${bytes} B`; + if (bytes < 1024 * 1024) return `${(bytes / 1024).toFixed(1)} KB`; + return `${(bytes / (1024 * 1024)).toFixed(1)} MB`; +} + +function useDebouncedValue(value: string, delayMs: number): string { + const [debounced, setDebounced] = useState(value); + useEffect(() => { + const timer = setTimeout(() => setDebounced(value), delayMs); + return () => clearTimeout(timer); + }, [value, delayMs]); + return debounced; +} + +function useLiveTail( + debouncedFilter: string, + levelFilter: string, + setLines: React.Dispatch> +) { + const [liveTail, setLiveTail] = useState(false); + const eventSourceRef = useRef(null); + + useEffect(() => { + if (!liveTail) { + if (eventSourceRef.current) { + eventSourceRef.current.close(); + eventSourceRef.current = null; + } + return; + } + + const es = createLogStream({ + filter: debouncedFilter || undefined, + level: levelFilter || undefined, + }); + + es.onmessage = (event: MessageEvent) => { + const newLines = (event.data as string).split('\n').filter(Boolean); + if (newLines.length > 0) { + setLines((prev) => { + const combined = [...prev, ...newLines]; + return combined.length > 5000 ? combined.slice(combined.length - 5000) : combined; + }); + } + }; + + es.onerror = () => { + logger.error('Log stream connection error'); + }; + + es.addEventListener('truncated', () => { + setLines([]); + }); + + eventSourceRef.current = es; + + return () => { + es.close(); + eventSourceRef.current = null; + }; + }, [liveTail, debouncedFilter, levelFilter, setLines]); + + return { liveTail, setLiveTail }; +} + +interface UseLogViewerResult { + lines: string[]; + isLoading: boolean; + error: string | null; + fileSize: number; + canGoOlder: boolean; + canGoNewer: boolean; + liveTail: boolean; + filterText: string; + levelFilter: string; + wrapLines: boolean; + pageSize: number; + expanded: boolean; + copyToastVisible: boolean; + logContainerRef: React.RefObject; + setFilterText: (v: string) => void; + handleLoadNewer: () => void; + handleLoadOlder: () => void; + handleLoadLatest: () => void; + handleToggleLiveTail: () => void; + handleToggleWrap: () => void; + handleToggleExpand: () => void; + handleCopyLine: (line: string) => void; + handlePageSizeChange: (e: React.ChangeEvent) => void; + handleScroll: () => void; + handleLevelChange: (e: React.ChangeEvent) => void; +} + +function useLogViewer(shouldLoad: boolean): UseLogViewerResult { + const [lines, setLines] = useState([]); + const [isLoading, setIsLoading] = useState(false); + const [error, setError] = useState(null); + const [fileSize, setFileSize] = useState(0); + const [backwardOffset, setBackwardOffset] = useState(0); + const [forwardOffset, setForwardOffset] = useState(0); + const [isAtLatest, setIsAtLatest] = useState(true); + const [filterText, setFilterText] = useState(''); + const [levelFilter, setLevelFilter] = useState(''); + const [wrapLines, setWrapLines] = useState(true); + const [pageSize, setPageSize] = useState(DEFAULT_PAGE_SIZE); + const [autoScroll, setAutoScroll] = useState(true); + const [expanded, setExpanded] = useState(false); + const [copyToastVisible, setCopyToastVisible] = useState(false); + const copyTimeoutRef = useRef | null>(null); + const logContainerRef = useRef(null); + + const debouncedFilter = useDebouncedValue(filterText, 300); + const { liveTail, setLiveTail } = useLiveTail(debouncedFilter, levelFilter, setLines); + + const scrollToBottom = useCallback(() => { + if (logContainerRef.current && autoScroll) { + logContainerRef.current.scrollTop = logContainerRef.current.scrollHeight; + } + }, [autoScroll]); + + const loadLogs = useCallback( + async (direction: 'forward' | 'backward', offset?: number) => { + setIsLoading(true); + setError(null); + try { + const response: LogResponse = await fetchLogs({ + offset, + limit: pageSize, + direction, + filter: debouncedFilter || undefined, + level: levelFilter || undefined, + }); + setLines(response.lines); + setFileSize(response.file_size); + + if (direction === 'backward') { + setBackwardOffset(response.next_offset); + setForwardOffset(offset ?? response.file_size); + setIsAtLatest(offset === undefined || offset >= response.file_size); + } else { + setForwardOffset(response.next_offset); + setBackwardOffset(offset ?? 0); + setIsAtLatest(!response.has_more); + } + } catch (err) { + const message = err instanceof Error ? err.message : 'Failed to load logs'; + logger.error('Failed to load logs:', err); + setError(message); + } finally { + setIsLoading(false); + } + }, + [debouncedFilter, levelFilter, pageSize] + ); + + useEffect(() => { + if (shouldLoad) { + loadLogs('backward'); + } + }, [loadLogs, shouldLoad]); + + useEffect(() => { + scrollToBottom(); + }, [lines, scrollToBottom]); + + return { + lines, + isLoading, + error, + fileSize, + canGoOlder: backwardOffset > 0, + canGoNewer: !isAtLatest, + liveTail, + filterText, + levelFilter, + wrapLines, + pageSize, + expanded, + copyToastVisible, + logContainerRef, + setFilterText, + handleLoadNewer: useCallback(() => { + if (forwardOffset < fileSize) loadLogs('forward', forwardOffset); + }, [loadLogs, forwardOffset, fileSize]), + handleLoadOlder: useCallback(() => { + if (backwardOffset > 0) loadLogs('backward', backwardOffset); + }, [loadLogs, backwardOffset]), + handleLoadLatest: useCallback(() => loadLogs('backward'), [loadLogs]), + handleToggleLiveTail: useCallback(() => { + if (!liveTail) { + loadLogs('backward').then(() => { + setLiveTail(true); + setAutoScroll(true); + }); + } else { + setLiveTail(false); + } + }, [liveTail, loadLogs, setLiveTail]), + handleToggleWrap: useCallback(() => setWrapLines((prev) => !prev), []), + handleToggleExpand: useCallback(() => setExpanded((prev) => !prev), []), + handleCopyLine: useCallback((line: string) => { + navigator.clipboard + .writeText(line) + .then(() => { + if (copyTimeoutRef.current) clearTimeout(copyTimeoutRef.current); + flushSync(() => setCopyToastVisible(true)); + copyTimeoutRef.current = setTimeout(() => setCopyToastVisible(false), 1500); + }) + .catch((err) => { + logger.error('Failed to copy log line:', err); + }); + }, []), + handlePageSizeChange: useCallback( + (e: React.ChangeEvent) => setPageSize(Number(e.target.value)), + [] + ), + handleScroll: useCallback(() => { + const container = logContainerRef.current; + if (!container) return; + const nearBottom = container.scrollHeight - container.scrollTop - container.clientHeight < 50; + setAutoScroll(nearBottom); + }, []), + handleLevelChange: useCallback( + (e: React.ChangeEvent) => setLevelFilter(e.target.value), + [] + ), + }; +} + +const LogsToolbar: React.FC<{ lv: UseLogViewerResult }> = ({ lv }) => ( + + lv.setFilterText(e.target.value)} + data-testid="logs-filter-input" + /> + + + + + + + + + + + + + + + + + + +); + +const LogsPagination: React.FC<{ lv: UseLogViewerResult }> = ({ lv }) => ( + +
+ + + +
+ + {lv.lines.length} lines + {lv.fileSize > 0 && ` \u2022 ${formatFileSize(lv.fileSize)}`} + + +); + +const LogsView: React.FC = () => { + const { role, isAdmin } = usePermissions(); + const admin = isAdmin(); + const lv = useLogViewer(admin); + + if (!admin) { + return ( + + + + + +
+ Logs + Role: {role ?? 'unknown'} +
+ + + Admin role required to view logs. + + + + + ); + } + + return ( + + + + + +
+ Logs + + Server log viewer{lv.fileSize > 0 && ` \u2022 ${formatFileSize(lv.fileSize)}`} + +
+ {lv.liveTail ? 'Live' : 'Paused'} + + + + + {lv.error && {lv.error}} + + + + + {lv.lines.length === 0 && !lv.isLoading && ( + No log lines to display. + )} + {lv.lines.map((line, i) => ( + lv.handleCopyLine(line)} + title="Click to copy" + > + {line} + + ))} + + + + + + + Copied to clipboard + + ); +}; + +export default LogsView; diff --git a/ui/src/views/admin/AdminNav.tsx b/ui/src/views/admin/AdminNav.tsx index 48791ebf..3b642f25 100644 --- a/ui/src/views/admin/AdminNav.tsx +++ b/ui/src/views/admin/AdminNav.tsx @@ -15,6 +15,9 @@ const AdminNav: React.FC = () => { (isActive ? 'active' : '')}> Tokens + (isActive ? 'active' : '')}> + Logs + ); };