use crate::protocol::{ProcessInfo, Stream};
use crate::tui::disk_log_reader::DiskLogReader;
use std::collections::{HashMap, VecDeque};
const MAX_BUFFER_LINES: usize = 10_000;
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum StreamMode {
Stdout,
Stderr,
Both,
}
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum LineSource {
Stdout,
Stderr,
}
pub struct OutputBuffer {
lines: VecDeque<(LineSource, String)>,
max_lines: usize,
stdout_count: usize,
stderr_count: usize,
}
impl OutputBuffer {
pub fn new(max_lines: usize) -> Self {
Self {
lines: VecDeque::with_capacity(max_lines),
max_lines,
stdout_count: 0,
stderr_count: 0,
}
}
pub fn push(&mut self, source: LineSource, line: String) {
if self.lines.len() == self.max_lines
&& let Some((evicted, _)) = self.lines.pop_front()
{
match evicted {
LineSource::Stdout => self.stdout_count -= 1,
LineSource::Stderr => self.stderr_count -= 1,
}
}
match source {
LineSource::Stdout => self.stdout_count += 1,
LineSource::Stderr => self.stderr_count += 1,
}
self.lines.push_back((source, line));
}
pub fn len(&self) -> usize {
self.lines.len()
}
pub fn is_empty(&self) -> bool {
self.lines.is_empty()
}
pub fn stdout_count(&self) -> usize {
self.stdout_count
}
pub fn stderr_count(&self) -> usize {
self.stderr_count
}
pub fn stdout_lines(&self) -> impl Iterator<Item = &str> {
self.lines
.iter()
.filter(|(src, _)| *src == LineSource::Stdout)
.map(|(_, s)| s.as_str())
}
pub fn stderr_lines(&self) -> impl Iterator<Item = &str> {
self.lines
.iter()
.filter(|(src, _)| *src == LineSource::Stderr)
.map(|(_, s)| s.as_str())
}
pub fn all_lines(&self) -> impl Iterator<Item = (LineSource, &str)> {
self.lines.iter().map(|(src, s)| (*src, s.as_str()))
}
}
pub struct FilteredIndex {
pub filter: String,
pub stream_mode: StreamMode,
pub matching_lines: Vec<usize>,
pub scanned_up_to: usize,
}
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum InputMode {
Normal,
FilterInput,
}
pub struct App {
pub processes: Vec<ProcessInfo>,
pub selected: usize,
pub buffers: HashMap<String, OutputBuffer>,
pub stream_mode: StreamMode,
pub paused: bool,
pub scroll_offsets: HashMap<String, usize>,
pub running: bool,
pub stop_all_on_quit: bool,
pub input_mode: InputMode,
pub filter_buf: String,
pub filter: Option<String>,
pub visible_height: usize,
pub disk_readers: HashMap<String, DiskLogReader>,
pub filtered_indices: HashMap<String, FilteredIndex>,
}
impl Default for App {
fn default() -> Self {
Self::new()
}
}
impl App {
pub fn new() -> Self {
Self {
processes: Vec::new(),
selected: 0,
buffers: HashMap::new(),
stream_mode: StreamMode::Stdout,
paused: false,
scroll_offsets: HashMap::new(),
running: true,
stop_all_on_quit: false,
input_mode: InputMode::Normal,
filter_buf: String::new(),
filter: None,
visible_height: 20,
disk_readers: HashMap::new(),
filtered_indices: HashMap::new(),
}
}
pub fn update_processes(&mut self, processes: Vec<ProcessInfo>) {
self.processes = processes;
if self.selected >= self.processes.len() && !self.processes.is_empty() {
self.selected = self.processes.len() - 1;
}
}
pub fn selected_name(&self) -> Option<&str> {
self.processes.get(self.selected).map(|p| p.name.as_str())
}
pub fn select_next(&mut self) {
if !self.processes.is_empty() {
self.selected = (self.selected + 1) % self.processes.len();
}
}
pub fn select_prev(&mut self) {
if !self.processes.is_empty() {
self.selected = if self.selected == 0 {
self.processes.len() - 1
} else {
self.selected - 1
};
}
}
pub fn cycle_stream_mode(&mut self) {
self.stream_mode = match self.stream_mode {
StreamMode::Stdout => StreamMode::Stderr,
StreamMode::Stderr => StreamMode::Both,
StreamMode::Both => StreamMode::Stdout,
};
if self.filter.is_some() {
let names: Vec<String> = self.disk_readers.keys().cloned().collect();
for name in names {
self.build_filtered_index(&name);
}
}
}
pub fn toggle_pause(&mut self) {
self.paused = !self.paused;
if !self.paused {
if let Some(name) = self.processes.get(self.selected).map(|p| p.name.clone()) {
self.scroll_offsets.remove(&name);
}
}
}
pub fn scroll_up_by(&mut self, lines: usize) {
if !self.paused {
self.paused = true;
}
if let Some(name) = self.selected_name().map(str::to_string) {
let max_offset = self
.line_count_for(&name)
.saturating_sub(self.visible_height);
let offset = self.scroll_offsets.entry(name).or_insert(0);
*offset = offset.saturating_add(lines).min(max_offset);
}
}
pub fn scroll_up(&mut self) {
let half_page = (self.visible_height / 2).max(1);
self.scroll_up_by(half_page);
}
pub fn scroll_down_by(&mut self, lines: usize) {
if let Some(name) = self.selected_name().map(str::to_string) {
let offset = self.scroll_offsets.entry(name).or_insert(0);
*offset = offset.saturating_sub(lines);
if *offset == 0 {
self.paused = false;
}
}
}
pub fn scroll_down(&mut self) {
let half_page = (self.visible_height / 2).max(1);
self.scroll_down_by(half_page);
}
pub fn scroll_to_top(&mut self) {
if !self.paused {
self.paused = true;
}
if let Some(name) = self.selected_name().map(str::to_string) {
let total = self.line_count_for(&name);
self.scroll_offsets.insert(name, total);
}
}
pub fn scroll_to_bottom(&mut self) {
if let Some(name) = self.selected_name().map(str::to_string) {
self.scroll_offsets.remove(&name);
self.paused = false;
}
}
fn line_count_for(&mut self, name: &str) -> usize {
if self.filter.is_some() {
return self
.filtered_indices
.get(name)
.map_or(0, |fi| fi.matching_lines.len());
}
self.total_line_count(name)
}
fn total_line_count(&mut self, name: &str) -> usize {
let hot = self.hot_line_count(name, None);
let disk = self.disk_line_count(name);
disk.max(hot)
}
fn disk_line_count(&mut self, name: &str) -> usize {
let mode = self.stream_mode;
self.disk_readers.get_mut(name).map_or(0, |r| match mode {
StreamMode::Stdout => r.line_count(LineSource::Stdout),
StreamMode::Stderr => r.line_count(LineSource::Stderr),
StreamMode::Both => r.line_count_both(),
})
}
fn hot_line_count(&self, name: &str, filter: Option<&str>) -> usize {
let Some(buf) = self.buffers.get(name) else {
return 0;
};
if let Some(pat) = filter {
let matches = |line: &str| line.contains(pat);
return match self.stream_mode {
StreamMode::Stdout => buf.stdout_lines().filter(|l| matches(l)).count(),
StreamMode::Stderr => buf.stderr_lines().filter(|l| matches(l)).count(),
StreamMode::Both => buf.all_lines().filter(|(_, l)| matches(l)).count(),
};
}
match self.stream_mode {
StreamMode::Stdout => buf.stdout_count(),
StreamMode::Stderr => buf.stderr_count(),
StreamMode::Both => buf.len(),
}
}
pub fn visible_lines(
&mut self,
name: &str,
visible_height: usize,
) -> Option<Vec<(LineSource, String)>> {
if self.filter.is_some() {
return Some(self.filtered_visible_lines(name, visible_height));
}
let total = self.total_line_count(name);
let scroll_offset = if self.paused {
self.scroll_offsets.get(name).copied().unwrap_or(0)
} else {
0
};
let window_end = total.saturating_sub(scroll_offset);
let window_start = window_end.saturating_sub(visible_height);
let count = window_end - window_start;
if count == 0 {
return Some(Vec::new());
}
let hot_len = self.hot_line_count(name, None);
let disk_count = self.disk_line_count(name);
let disk_boundary = disk_count.saturating_sub(hot_len);
if window_start >= disk_boundary {
let hot_start = window_start - disk_boundary;
Some(self.hot_buffer_range(name, hot_start, count))
} else if window_end <= disk_boundary {
Some(self.disk_read_range(name, window_start, count))
} else {
let disk_part = disk_boundary - window_start;
let hot_part = window_end - disk_boundary;
let mut lines = self.disk_read_range(name, window_start, disk_part);
lines.extend(self.hot_buffer_range(name, 0, hot_part));
Some(lines)
}
}
fn hot_buffer_range(
&self,
name: &str,
start: usize,
count: usize,
) -> Vec<(LineSource, String)> {
let Some(buf) = self.buffers.get(name) else {
return Vec::new();
};
match self.stream_mode {
StreamMode::Stdout => buf
.stdout_lines()
.skip(start)
.take(count)
.map(|l| (LineSource::Stdout, l.to_string()))
.collect(),
StreamMode::Stderr => buf
.stderr_lines()
.skip(start)
.take(count)
.map(|l| (LineSource::Stderr, l.to_string()))
.collect(),
StreamMode::Both => buf
.all_lines()
.skip(start)
.take(count)
.map(|(src, l)| (src, l.to_string()))
.collect(),
}
}
fn disk_read_range(
&mut self,
name: &str,
start: usize,
count: usize,
) -> Vec<(LineSource, String)> {
let Some(reader) = self.disk_readers.get_mut(name) else {
return Vec::new();
};
match self.stream_mode {
StreamMode::Stdout => reader
.read_lines(LineSource::Stdout, start, count)
.unwrap_or_default()
.into_iter()
.map(|l| (LineSource::Stdout, l))
.collect(),
StreamMode::Stderr => reader
.read_lines(LineSource::Stderr, start, count)
.unwrap_or_default()
.into_iter()
.map(|l| (LineSource::Stderr, l))
.collect(),
StreamMode::Both => reader.read_interleaved(start, count).unwrap_or_default(),
}
}
fn filtered_visible_lines(
&mut self,
name: &str,
visible_height: usize,
) -> Vec<(LineSource, String)> {
let total = self
.filtered_indices
.get(name)
.map_or(0, |fi| fi.matching_lines.len());
if total == 0 {
return Vec::new();
}
let scroll_offset = if self.paused {
self.scroll_offsets.get(name).copied().unwrap_or(0)
} else {
0
};
let window_end = total.saturating_sub(scroll_offset);
let window_start = window_end.saturating_sub(visible_height);
let count = window_end - window_start;
if count == 0 {
return Vec::new();
}
let line_numbers: Vec<usize> = self
.filtered_indices
.get(name)
.map(|fi| fi.matching_lines[window_start..window_end].to_vec())
.unwrap_or_default();
let mode = self.stream_mode;
if let Some(reader) = self.disk_readers.get_mut(name) {
reader.read_scattered_lines(mode, &line_numbers)
} else {
Vec::new()
}
}
fn build_filtered_index(&mut self, name: &str) {
let Some(filter) = self.filter.clone() else {
return;
};
let mode = self.stream_mode;
let matching_lines = if let Some(reader) = self.disk_readers.get_mut(name) {
reader.scan_matching_lines(&filter, mode)
} else {
Vec::new()
};
let scanned_up_to = match mode {
StreamMode::Stdout => self
.disk_readers
.get_mut(name)
.map_or(0, |r| r.line_count(LineSource::Stdout)),
StreamMode::Stderr => self
.disk_readers
.get_mut(name)
.map_or(0, |r| r.line_count(LineSource::Stderr)),
StreamMode::Both => self
.disk_readers
.get_mut(name)
.map_or(0, DiskLogReader::line_count_both),
};
self.filtered_indices.insert(
name.to_string(),
FilteredIndex {
filter,
stream_mode: mode,
matching_lines,
scanned_up_to,
},
);
}
pub fn start_filter(&mut self) {
self.input_mode = InputMode::FilterInput;
self.filter_buf = self.filter.clone().unwrap_or_default();
}
pub fn confirm_filter(&mut self) {
self.input_mode = InputMode::Normal;
if self.filter_buf.is_empty() {
self.filter = None;
self.filtered_indices.clear();
} else {
self.filter = Some(self.filter_buf.clone());
let names: Vec<String> = self.disk_readers.keys().cloned().collect();
for name in names {
self.build_filtered_index(&name);
}
}
}
pub fn cancel_filter(&mut self) {
self.input_mode = InputMode::Normal;
self.filter_buf.clear();
}
pub fn clear_filter(&mut self) {
self.filter = None;
self.filter_buf.clear();
self.filtered_indices.clear();
}
pub fn push_output(&mut self, process: &str, stream: Stream, line: &str) {
let buf = self
.buffers
.entry(process.to_string())
.or_insert_with(|| OutputBuffer::new(MAX_BUFFER_LINES));
let source = match stream {
Stream::Stdout => LineSource::Stdout,
Stream::Stderr => LineSource::Stderr,
};
buf.push(source, line.to_string());
if self.filter.is_some() {
self.update_filtered_index(process);
}
}
fn update_filtered_index(&mut self, process: &str) {
let Some(fi) = self.filtered_indices.get(process) else {
return;
};
let filter = fi.filter.clone();
let mode = fi.stream_mode;
let scanned_up_to = fi.scanned_up_to;
let reader = match self.disk_readers.get_mut(process) {
Some(r) => r,
None => return,
};
let current_total = match mode {
StreamMode::Stdout => reader.line_count(LineSource::Stdout),
StreamMode::Stderr => reader.line_count(LineSource::Stderr),
StreamMode::Both => reader.line_count_both(),
};
if current_total <= scanned_up_to {
return;
}
let new_matches = reader.scan_matching_lines_from(&filter, mode, scanned_up_to);
let fi = self.filtered_indices.get_mut(process).unwrap();
fi.matching_lines.extend(new_matches);
fi.scanned_up_to = current_total;
}
pub fn quit(&mut self) {
self.running = false;
}
pub fn quit_and_stop(&mut self) {
self.stop_all_on_quit = true;
self.running = false;
}
pub fn running_count(&self) -> usize {
self.processes
.iter()
.filter(|p| p.state == crate::protocol::ProcessState::Running)
.count()
}
pub fn exited_count(&self) -> usize {
self.processes
.iter()
.filter(|p| p.state == crate::protocol::ProcessState::Exited)
.count()
}
pub fn failed_count(&self) -> usize {
self.processes
.iter()
.filter(|p| p.state == crate::protocol::ProcessState::Failed)
.count()
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::protocol::{ProcessInfo, ProcessState, Stream};
fn make_process(name: &str, state: ProcessState) -> ProcessInfo {
let exit_code = if state == ProcessState::Exited {
Some(0)
} else {
None
};
let uptime_secs = if state == ProcessState::Running {
Some(10)
} else {
None
};
ProcessInfo {
name: name.into(),
id: format!("p-{}", name),
pid: 100,
state,
exit_code,
uptime_secs,
command: "true".into(),
port: None,
url: None,
restart_count: None,
max_restarts: None,
restart_policy: None,
watched: None,
}
}
#[test]
fn test_select_next_wraps() {
let mut app = App::new();
app.update_processes(vec![
make_process("a", ProcessState::Running),
make_process("b", ProcessState::Running),
make_process("c", ProcessState::Running),
]);
app.selected = 2; app.select_next();
assert_eq!(app.selected, 0);
}
#[test]
fn test_select_prev_wraps() {
let mut app = App::new();
app.update_processes(vec![
make_process("a", ProcessState::Running),
make_process("b", ProcessState::Running),
make_process("c", ProcessState::Running),
]);
app.selected = 0;
app.select_prev();
assert_eq!(app.selected, 2);
}
#[test]
fn test_cycle_stream_mode() {
let mut app = App::new();
assert_eq!(app.stream_mode, StreamMode::Stdout);
app.cycle_stream_mode();
assert_eq!(app.stream_mode, StreamMode::Stderr);
app.cycle_stream_mode();
assert_eq!(app.stream_mode, StreamMode::Both);
app.cycle_stream_mode();
assert_eq!(app.stream_mode, StreamMode::Stdout);
}
#[test]
fn test_toggle_pause() {
let mut app = App::new();
assert!(!app.paused);
app.toggle_pause();
assert!(app.paused);
app.toggle_pause();
assert!(!app.paused);
}
#[test]
fn test_push_output() {
let mut app = App::new();
app.push_output("web", Stream::Stdout, "hello world");
let buf = app.buffers.get("web").unwrap();
assert_eq!(buf.stdout_lines().count(), 1);
assert_eq!(buf.stdout_lines().next().unwrap(), "hello world");
}
#[test]
fn test_running_count() {
let mut app = App::new();
app.update_processes(vec![
make_process("a", ProcessState::Running),
make_process("b", ProcessState::Exited),
make_process("c", ProcessState::Running),
]);
assert_eq!(app.running_count(), 2);
}
#[test]
fn test_exited_count() {
let mut app = App::new();
app.update_processes(vec![
make_process("a", ProcessState::Running),
make_process("b", ProcessState::Exited),
make_process("c", ProcessState::Exited),
]);
assert_eq!(app.exited_count(), 2);
}
#[test]
fn test_output_buffer_counters() {
let mut buf = OutputBuffer::new(5);
assert_eq!(buf.len(), 0);
assert_eq!(buf.stdout_count(), 0);
assert_eq!(buf.stderr_count(), 0);
assert!(buf.is_empty());
buf.push(LineSource::Stdout, "a".into());
buf.push(LineSource::Stderr, "b".into());
buf.push(LineSource::Stdout, "c".into());
assert_eq!(buf.len(), 3);
assert_eq!(buf.stdout_count(), 2);
assert_eq!(buf.stderr_count(), 1);
buf.push(LineSource::Stdout, "d".into());
buf.push(LineSource::Stderr, "e".into());
assert_eq!(buf.len(), 5);
assert_eq!(buf.stdout_count(), 3);
assert_eq!(buf.stderr_count(), 2);
buf.push(LineSource::Stderr, "f".into());
assert_eq!(buf.len(), 5);
assert_eq!(buf.stdout_count(), 2);
assert_eq!(buf.stderr_count(), 3);
}
#[test]
fn test_visible_lines_hot_buffer_only() {
let mut app = App::new();
for i in 0..20 {
app.push_output("web", Stream::Stdout, &format!("line {}", i));
}
app.update_processes(vec![make_process("web", ProcessState::Running)]);
app.visible_height = 10;
let lines = app.visible_lines("web", 10).unwrap();
assert_eq!(lines.len(), 10);
assert_eq!(lines[0].1, "line 10");
assert_eq!(lines[9].1, "line 19");
}
#[test]
fn test_visible_lines_paused_scrolled() {
let mut app = App::new();
for i in 0..50 {
app.push_output("web", Stream::Stdout, &format!("line {}", i));
}
app.update_processes(vec![make_process("web", ProcessState::Running)]);
app.visible_height = 10;
app.paused = true;
app.scroll_offsets.insert("web".into(), 20);
let lines = app.visible_lines("web", 10).unwrap();
assert_eq!(lines.len(), 10);
assert_eq!(lines[0].1, "line 20");
assert_eq!(lines[9].1, "line 29");
}
#[test]
fn test_visible_lines_with_disk_reader() {
use crate::daemon::log_index::{IndexRecord, IndexWriter, idx_path_for};
use crate::tui::disk_log_reader::DiskLogReader;
let dir = tempfile::tempdir().unwrap();
let log_path = dir.path().join("web.stdout");
let idx_path = idx_path_for(&log_path);
let mut log_content = String::new();
let mut writer = IndexWriter::create(&idx_path, 0).unwrap();
let mut offset = 0u64;
for i in 0..100 {
let line = format!("disk line {}", i);
writer
.append(IndexRecord {
byte_offset: offset,
seq: i,
})
.unwrap();
log_content.push_str(&line);
log_content.push('\n');
offset += line.len() as u64 + 1;
}
writer.flush().unwrap();
std::fs::write(&log_path, log_content).unwrap();
let mut app = App::new();
app.disk_readers.insert(
"web".into(),
DiskLogReader::new(dir.path().to_path_buf(), "web".into()),
);
for i in 90..100 {
app.push_output("web", Stream::Stdout, &format!("disk line {}", i));
}
app.update_processes(vec![make_process("web", ProcessState::Running)]);
app.visible_height = 10;
assert_eq!(app.total_line_count("web"), 100);
let lines = app.visible_lines("web", 10).unwrap();
assert_eq!(lines.len(), 10);
assert_eq!(lines[0].1, "disk line 90");
assert_eq!(lines[9].1, "disk line 99");
app.paused = true;
app.scroll_offsets.insert("web".into(), 90);
let lines = app.visible_lines("web", 10).unwrap();
assert_eq!(lines.len(), 10);
assert_eq!(lines[0].1, "disk line 0");
assert_eq!(lines[9].1, "disk line 9");
app.scroll_offsets.insert("web".into(), 5);
let lines = app.visible_lines("web", 10).unwrap();
assert_eq!(lines.len(), 10);
assert_eq!(lines[0].1, "disk line 85");
assert_eq!(lines[4].1, "disk line 89");
assert_eq!(lines[5].1, "disk line 90");
assert_eq!(lines[9].1, "disk line 94");
}
#[test]
fn test_visible_lines_with_filter_uses_filtered_index() {
let mut app = App::new();
app.push_output("web", Stream::Stdout, "hello");
app.filter = Some("hello".into());
let lines = app.visible_lines("web", 10).unwrap();
assert!(lines.is_empty());
}
#[test]
fn test_hot_line_count_with_and_without_filter() {
let mut app = App::new();
app.push_output("web", Stream::Stdout, "hello world");
app.push_output("web", Stream::Stdout, "goodbye world");
app.push_output("web", Stream::Stdout, "hello again");
assert_eq!(app.hot_line_count("web", None), 3);
assert_eq!(app.hot_line_count("web", Some("hello")), 2);
assert_eq!(app.hot_line_count("web", Some("xyz")), 0);
}
}