diff --git a/Cargo.lock b/Cargo.lock index 5e845c5..f4757f3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -623,6 +623,12 @@ version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dd2e7510819d6fbf51a5545c8f922716ecfb14df168a3242f7d33e0239efe6a1" +[[package]] +name = "fastrand" +version = "2.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9f1f227452a390804cdb637b74a86990f2a7d7ba4b7d5693aac9b4dd6defd8d6" + [[package]] name = "filedescriptor" version = "0.8.3" @@ -1676,6 +1682,19 @@ dependencies = [ "syn 2.0.118", ] +[[package]] +name = "serde_json" +version = "1.0.150" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e8014e44b4736ed0538adeecded0fce2a272f22dc9578a7eb6b2d9993c74cfb9" +dependencies = [ + "itoa", + "memchr", + "serde", + "serde_core", + "zmij", +] + [[package]] name = "sha2" version = "0.10.9" @@ -1850,10 +1869,26 @@ dependencies = [ "regex", "rustls", "rustls-pemfile", + "serde", + "serde_json", + "tempfile", "tokio", "tokio-rustls", ] +[[package]] +name = "tempfile" +version = "3.27.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" +dependencies = [ + "fastrand", + "getrandom 0.4.3", + "once_cell", + "rustix", + "windows-sys 0.61.2", +] + [[package]] name = "termina" version = "0.3.3" @@ -2494,3 +2529,9 @@ name = "zeroize" version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e13c156562582aa81c60cb29407084cdb54c4164760106ab78e6c5b0858cf64e" + +[[package]] +name = "zmij" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b" diff --git a/Cargo.toml b/Cargo.toml index 0e7a569..66df9ec 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -20,6 +20,8 @@ bytes = "1" clap = { version = "4", features = ["derive"] } chrono = { version = "0.4", default-features = false, features = ["clock"] } regex = "1" +serde = { version = "1", features = ["derive"] } +serde_json = "1" # --- TUI (`--tui`) --- # ratatui re-exports crossterm (its default backend), so we use that for events. @@ -50,3 +52,4 @@ rcgen = "0.14" # is dev-only: it never reaches a normal `cargo build`/release/`cargo install`. [dev-dependencies] clap_mangen = "0.2" +tempfile = "3" diff --git a/README.md b/README.md index d2a6616..b973324 100644 --- a/README.md +++ b/README.md @@ -27,6 +27,8 @@ tapgres -p 5432 -i eth0 # capture a specific interface tapgres --mode mitm \ # decode an encrypted session via the proxy --listen 127.0.0.1:15432 --upstream 127.0.0.1:5432 tapgres --tui -Y 'message.type == "Query"' # interactive view, filtered +tapgres --save session.jsonl # capture and tee every record to disk +tapgres --replay session.jsonl --tui # reopen it without live capture ``` For making a client trust the mitm proxy's auto-generated CA, see @@ -60,12 +62,42 @@ sudo setcap cap_net_raw+ep $(which tapgres) | `w` / `r` | wrap / rich display | | `c` | clear | | `y` | edit the display filter | +| `/` / `:` | command bar (`:save FILE`, `:open FILE`) | | `Esc` | clear the display filter | Display filters (`-Y` / `--display-filter`) use a small typed expression language with fields like `message.type`, `message.text`, `client.ip`, and `client.port`. See `man tapgres` for the full field and operator reference. +## Save and replay + +`--save FILE` continuously writes every output record to versioned JSONL while +stdout or the TUI continues normally. Recording happens before display +filtering and before the TUI's 50,000-record history cap, so hidden or evicted +live records are still saved. An existing destination is replaced. + +`--replay FILE` uses a saved session instead of pcap/mitm capture. Replay is +instant, preserves the original timestamps and structured rich-view data, and +passes through the same display filters and renderers as live traffic: + +```sh +tapgres --replay session.jsonl +tapgres --replay session.jsonl --tui --tui-rich +tapgres --replay session.jsonl -Y 'message.type == "Query"' +``` + +In the TUI, `/` or `:` opens the command bar. `:save FILE` (also `:w`) writes +the currently retained events and then continuously records future traffic. +If older events have already left the TUI history, the footer reports the +omission. `:open FILE` (also `:o`) validates the complete file, replaces the +current view with its newest 50,000 records, and switches the session to replay +mode. It closes any active recorder, and subsequent live-source records are +discarded so live and replayed timelines never mix. + +Schema version 1 and its compatibility rules are defined in +[`docs/session-format.md`](docs/session-format.md). Unsupported schema versions +and malformed records are refused with a file and line-numbered error. + ## Installation **Prebuilt binary** (Linux x86_64, from diff --git a/docs/session-format.md b/docs/session-format.md new file mode 100644 index 0000000..8767a74 --- /dev/null +++ b/docs/session-format.md @@ -0,0 +1,82 @@ +# Tapgres saved-session format + +Tapgres saves sessions as UTF-8 JSON Lines (JSONL): one complete JSON object per +line. Version 1 is designed to round-trip every `decode::Output` variant through +the same display-filter and rendering pipeline used by live capture. + +The format is local and stream-oriented. `--save` and `:save` replace an +existing destination, then record events before display filtering or TUI +history eviction. Files may contain sensitive SQL, credentials, errors, and +returned row values and should be protected accordingly. + +## Common fields + +Every line contains: + +| Field | Type | Meaning | +| --- | --- | --- | +| `schema_version` | unsigned integer | Version of this record shape; currently `1`. | +| `timestamp` | RFC 3339 string | Original message capture time, or record time for operational lines/status. | +| `record_type` | string | `message`, `line`, or `status`. | + +Blank lines are ignored. A malformed record aborts replay with the file path and +line number. + +## Message records + +A decoded PostgreSQL message contains the structured values required by display +filters and rich rendering: + +```json +{"schema_version":1,"timestamp":"2026-07-17T12:34:56.789+01:00","record_type":"message","direction":"f2b","message_type":"Query","text":"SELECT * FROM orders","rendered":"[12:34:56.789] [F→B] Query: SELECT * FROM orders","client":"127.0.0.1:40005"} +``` + +| Field | Type | Meaning | +| --- | --- | --- | +| `direction` | string | `f2b` for frontend/client to backend/server, or `b2f` for the reverse. | +| `message_type` | string | Decoded pgwire message name, such as `Query` or `DataRow`. | +| `text` | string | Decoded message body used by `message.text` filters. | +| `rendered` | string | Stable flat terminal representation, including the original display timestamp. | +| `client` | socket-address string | Owning connection's client IP and port. | +| `detail` | object, optional | Structured rich-rendering payload described below. | + +### Rich detail + +`RowDescription` preserves field names, PostgreSQL type OIDs, and format codes: + +```json +{"detail_type":"row_description","columns":[{"name":"id","type_oid":23,"format_code":0}]} +``` + +`DataRow` preserves its already-decoded display value alongside the cached +column name and type OID: + +```json +{"detail_type":"data_row","columns":[{"name":"id","type_oid":23,"value":"'1'"}]} +``` + +The detail object is nested under the message record's `detail` field. Keeping +both forms allows replay to reproduce rich tables without re-decoding pgwire +bytes while retaining the stable flat transcript. + +## Operational records + +Connection/capture lines and status records retain their original text: + +```json +{"schema_version":1,"timestamp":"2026-07-17T12:34:56.789+01:00","record_type":"line","text":"=== new connection 127.0.0.1:40005 -> 127.0.0.1:5432 ==="} +{"schema_version":1,"timestamp":"2026-07-17T12:34:56.790+01:00","record_type":"status","text":"tapgres: capturing on 'lo'"} +``` + +They remain outside display filtering, matching live behavior. + +## Compatibility policy + +- Tapgres writes only the current schema version. +- Version 1 readers require `schema_version: 1` on every non-blank line. +- Unknown older or newer versions are refused; there is no silent best-effort + conversion. +- A future incompatible shape must increment `schema_version` and provide an + explicit migration path if backward compatibility is desired. +- Replay preserves recorded `rendered` output rather than reformatting it, while + filters and rich mode use the structured fields. diff --git a/flake.nix b/flake.nix index 0c103e9..0b83531 100644 --- a/flake.nix +++ b/flake.nix @@ -38,15 +38,15 @@ # Source cleaning: keep cargo's own selection (crane's # commonCargoSources — every .rs/.toml/Cargo.lock across the workspace), - # plus the committed `man/sections.md` that the gen_manpage example - # embeds with include_str!. cleanCargoSource alone strips it (cargo - # doesn't track it), which breaks the example's build; fileset.toSource - # makes the extra include explicit and unambiguous. + # plus committed non-Rust inputs used by builds and tests. Cargo does + # not track these files, so commonCargoSources strips them unless the + # fileset includes them explicitly. src = pkgs.lib.fileset.toSource { root = ./.; fileset = pkgs.lib.fileset.unions [ (craneLib.fileset.commonCargoSources ./.) ./man/sections.md + ./tests/fixtures/session-v1.jsonl ]; }; diff --git a/man/sections.md b/man/sections.md index b0ed48d..91fa463 100644 --- a/man/sections.md +++ b/man/sections.md @@ -1,3 +1,32 @@ +# SAVED SESSIONS + +`--save FILE` continuously writes every output record as versioned JSONL while +normal stdout or TUI rendering continues. Recording occurs before display +filtering and before the TUI history cap, so hidden and evicted live records +remain in the saved session. Operational line/status records are saved along +with decoded PostgreSQL messages. An existing destination file is replaced. + +`--replay FILE` reads a saved session instead of starting pcap or mitm. Records +are replayed immediately through the same stdout/TUI renderer and display +filter as live traffic. Original capture timestamps, client address, direction, +message type/text, and rich RowDescription/DataRow details are preserved. A +replay can be copied to another file with `--save`; the input and output paths +must differ. + +The TUI command bar opens with `/` or `:`. `:save FILE` (`:w FILE`) writes the +currently retained history and continuously records future events. If earlier +events have left the 50,000-record TUI history, a footer warning reports the +omission. `:open FILE` (`:o FILE`) validates the complete file before replacing +the view, retains its newest 50,000 records, switches the UI to replay mode, and +closes any active recorder. Subsequent live-source display records are discarded +so timelines do not mix. + +The current on-disk schema version is 1. Each JSONL record carries its own +`schema_version` and RFC 3339 timestamp. Unknown versions and malformed records +are refused with the file path and line number; tapgres does not guess at an +incompatible shape. The full format is documented in `docs/session-format.md` +in the source repository. + # DISPLAY FILTER EXPRESSIONS The `-Y` / `--display-filter` option limits decoded PostgreSQL messages in diff --git a/src/cli.rs b/src/cli.rs index 1672c57..8f2d7f0 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -22,7 +22,8 @@ use crate::state; with the pgwire protocol layer. Use --mode pcap (the default) to passively \ capture a local port with libpcap (cleartext only), or --mode mitm to run a \ local TLS-terminating proxy that decrypts encrypted sessions. Add --tui to \ - either source for an interactive, scrollable, filterable view.", + either source for an interactive, scrollable, filterable view. Use --save to \ + record versioned JSONL or --replay to open a saved session without capture.", before_help = crate::tui::BANNER )] pub struct Args { @@ -30,7 +31,7 @@ pub struct Args { #[arg(long, value_enum, default_value_t = Mode::Pcap)] pub mode: Mode, - /// Interactive TUI instead of line-oriented stdout (works with any --mode). + /// Interactive TUI instead of line-oriented stdout (works with live or replay sources). #[arg(long, default_value_t = false)] pub tui: bool, @@ -45,6 +46,34 @@ pub struct Args { #[arg(short = 'Y', long = "display-filter")] pub display_filter: Option, + /// Save every live or replayed output record as versioned JSONL while + /// continuing to render normally. Recording happens before display + /// filtering and before the TUI history cap is applied. An existing file + /// is replaced. + #[arg(long, value_name = "FILE")] + pub save: Option, + + /// Read a saved JSONL session instead of starting pcap or mitm capture. + /// Replay is loaded at full speed and preserves original timestamps. + #[arg( + long, + value_name = "FILE", + conflicts_with_all = [ + "mode", + "port", + "interface", + "no_promisc", + "snaplen", + "listen", + "upstream", + "tls_dir", + "tls_cert", + "tls_key", + "no_upstream_tls" + ] + )] + pub replay: Option, + /// Maximum retained open + recently-closed connection records. /// Open connections are never evicted. #[arg(long, default_value_t = state::DEFAULT_CONNECTION_CAP)] @@ -120,3 +149,34 @@ pub enum Mode { pub fn command() -> clap::Command { Args::command() } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parses_save_and_replay_as_file_source_options() { + let args = Args::try_parse_from([ + "tapgres", + "--replay", + "capture.jsonl", + "--save", + "copy.jsonl", + "--tui", + ]) + .unwrap(); + + assert_eq!(args.replay, Some(PathBuf::from("capture.jsonl"))); + assert_eq!(args.save, Some(PathBuf::from("copy.jsonl"))); + assert!(args.tui); + } + + #[test] + fn replay_rejects_live_source_options() { + let error = + Args::try_parse_from(["tapgres", "--replay", "capture.jsonl", "--mode", "mitm"]) + .unwrap_err(); + + assert_eq!(error.kind(), clap::error::ErrorKind::ArgumentConflict); + } +} diff --git a/src/decode.rs b/src/decode.rs index 00381ed..941c437 100644 --- a/src/decode.rs +++ b/src/decode.rs @@ -12,7 +12,7 @@ use std::sync::Mutex; use crossbeam_channel::Sender; use bytes::{Buf, Bytes}; -use chrono::Local; +use chrono::{Local, SecondsFormat}; use crate::filter::{DisplayFilter, DisplayMessage, MessageDirection}; use crate::flow::{Direction, Role}; @@ -169,6 +169,13 @@ fn deliver(record: Output) { } } +/// Feed a previously decoded record through the active consumer. File replay +/// uses this entry point so it follows the exact same stdout/TUI path as live +/// capture without exposing the decoder's routing internals. +pub fn replay(record: Output) { + deliver(record); +} + /// Emit one decoded protocol line. Routed to the output consumer, or stdout if /// none is wired. pub fn out(line: String) { @@ -224,13 +231,17 @@ impl MessageEmitter { } fn emit_with_detail(self, kind: &str, text: &str, detail: Option) { + let captured_at = Local::now(); + let timestamp = captured_at.to_rfc3339_opts(SecondsFormat::Millis, true); + let display_time = captured_at.format("%H:%M:%S%.3f"); let rendered = if text.is_empty() { - format!("[{}] [{}] {}", ts(), dir_tag(self.role), kind) + format!("[{display_time}] [{}] {kind}", dir_tag(self.role)) } else { - format!("[{}] [{}] {}: {}", ts(), dir_tag(self.role), kind, text) + format!("[{display_time}] [{}] {kind}: {text}", dir_tag(self.role)) }; deliver(Output::Message { message: DisplayMessage { + timestamp, rendered, client: self.client, direction: if self.role == Role::Client { @@ -246,9 +257,13 @@ impl MessageEmitter { } fn warn(self, msg: &str) { - let rendered = format!("[{}] [{}] ⚠ {}", ts(), dir_tag(self.role), msg); + let captured_at = Local::now(); + let timestamp = captured_at.to_rfc3339_opts(SecondsFormat::Millis, true); + let display_time = captured_at.format("%H:%M:%S%.3f"); + let rendered = format!("[{display_time}] [{}] ⚠ {msg}", dir_tag(self.role)); deliver(Output::Message { message: DisplayMessage { + timestamp, rendered, client: self.client, direction: if self.role == Role::Client { @@ -1039,6 +1054,7 @@ mod tests { .unwrap(); let query = Output::Message { message: DisplayMessage { + timestamp: "2026-07-17T12:34:56.789+01:00".into(), rendered: "query".into(), client: "127.0.0.1:40005".parse().unwrap(), direction: MessageDirection::FrontendToBackend, @@ -1049,6 +1065,7 @@ mod tests { }; let row = Output::Message { message: DisplayMessage { + timestamp: "2026-07-17T12:34:56.789+01:00".into(), rendered: "row".into(), client: "127.0.0.1:40005".parse().unwrap(), direction: MessageDirection::BackendToFrontend, diff --git a/src/filter.rs b/src/filter.rs index cfe2450..07ff802 100644 --- a/src/filter.rs +++ b/src/filter.rs @@ -25,6 +25,9 @@ pub enum MessageDirection { /// A decoded message plus the structured fields used by display filters. #[derive(Clone, Debug)] pub struct DisplayMessage { + /// Original capture time in RFC 3339 with millisecond precision. Kept + /// separately from `rendered` so saved sessions preserve real timestamps. + pub timestamp: String, pub rendered: String, pub client: SocketAddr, pub direction: MessageDirection, @@ -672,6 +675,7 @@ mod tests { fn message() -> DisplayMessage { DisplayMessage { + timestamp: "2026-07-17T12:34:56.789+01:00".into(), rendered: "line".into(), client: "127.0.0.1:40005".parse().unwrap(), direction: MessageDirection::FrontendToBackend, diff --git a/src/lib.rs b/src/lib.rs index a5ce2c3..cb71959 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -10,5 +10,6 @@ pub mod filter; pub mod flow; pub mod net; pub mod proxy; +pub mod session; pub mod state; pub mod tui; diff --git a/src/main.rs b/src/main.rs index 0eef08c..b9ac19f 100644 --- a/src/main.rs +++ b/src/main.rs @@ -11,19 +11,24 @@ //! instead of the server; the proxy decrypts the client leg, decodes the //! traffic in the middle, and forwards it to the real server. See //! [`tapgres::proxy`]. +//! - `--replay FILE`: opens a versioned JSONL session instead of capturing and +//! feeds its decoded records through the same stdout or TUI renderer. //! //! Add `--tui` to either mode for an interactive, scrollable, filterable view //! instead of line-oriented stdout. See [`tapgres::tui`]. use std::error::Error; +use std::fs; use std::io::Write; +use std::io::{self}; +use std::path::Path; use std::path::PathBuf; use std::sync::Arc; use clap::Parser; use tapgres::cli::{Args, Mode}; -use tapgres::{capture, decode, filter::DisplayFilter, proxy, state, tui}; +use tapgres::{capture, decode, filter::DisplayFilter, proxy, session, state, tui}; fn main() -> Result<(), Box> { let args = Args::parse(); @@ -32,6 +37,18 @@ fn main() -> Result<(), Box> { args.conn_history, args.rate_history, )); + if let Some(replay) = args.replay { + if let Some(save) = &args.save { + ensure_distinct_files(&replay, save)?; + } + return if args.tui { + tui::run_replay(replay, metrics, args.tui_rich, filter, args.save) + } else { + run_stdout(filter, args.save, move || { + session::read_with(replay, decode::replay).map_err(Into::into) + }) + }; + } match args.mode { Mode::Pcap => { let opts = capture::PcapOpts { @@ -41,9 +58,9 @@ fn main() -> Result<(), Box> { snaplen: args.snaplen, }; if args.tui { - tui::run_pcap(opts, metrics, args.tui_rich, filter) + tui::run_pcap(opts, metrics, args.tui_rich, filter, args.save) } else { - run_stdout(filter, move || capture::run(opts, metrics)) + run_stdout(filter, args.save, move || capture::run(opts, metrics)) } } Mode::Mitm => { @@ -56,9 +73,9 @@ fn main() -> Result<(), Box> { no_upstream_tls: args.no_upstream_tls, }; if args.tui { - tui::run_mitm(opts, metrics, args.tui_rich, filter) + tui::run_mitm(opts, metrics, args.tui_rich, filter, args.save) } else { - run_stdout(filter, move || proxy::run(opts, metrics)) + run_stdout(filter, args.save, move || proxy::run(opts, metrics)) } } } @@ -67,18 +84,32 @@ fn main() -> Result<(), Box> { /// Run `source` with its decoded output funneled through a single consumer /// thread: decoded lines to stdout, status to stderr. When `source` returns, /// close the channel and join the consumer so nothing is left unflushed. -fn run_stdout(filter: DisplayFilter, source: F) -> Result<(), Box> +fn run_stdout( + filter: DisplayFilter, + save: Option, + source: F, +) -> Result<(), Box> where F: FnOnce() -> Result<(), Box>, { let (tx, rx) = crossbeam_channel::unbounded(); decode::set_output(tx); + let mut recorder = save.map(session::SessionWriter::create).transpose()?; let printer = std::thread::Builder::new() .name("tapgres-out".into()) - .spawn(move || { + .spawn(move || -> io::Result<()> { let mut stdout = std::io::stdout().lock(); let mut stderr = std::io::stderr().lock(); + let mut recorder_error = None; while let Ok(record) = rx.recv() { + if let Some(writer) = recorder.as_mut() { + if let Err(error) = writer.write(&record) { + let message = format!("tapgres: recording stopped: {error}"); + let _ = writeln!(stderr, "{message}"); + recorder_error = Some(io::Error::other(message)); + recorder = None; + } + } if !record.matches_filter(&filter) { continue; } @@ -94,13 +125,66 @@ where } } } + if let Some(writer) = recorder.as_mut() { + writer.flush().map_err(io::Error::other)?; + } let _ = stdout.flush(); let _ = stderr.flush(); + if let Some(error) = recorder_error { + return Err(error); + } + Ok(()) })?; let result = source(); decode::close_output(); - let _ = printer.join(); - result + let consumer_result = printer + .join() + .map_err(|_| io::Error::other("output consumer thread panicked"))?; + result?; + consumer_result?; + Ok(()) +} + +/// Prevent `--replay FILE --save FILE` from truncating the input before it is +/// read. Canonicalising the existing input and the output's parent also catches +/// equivalent relative paths and symlinked directories. +fn ensure_distinct_files(input: &Path, output: &Path) -> Result<(), Box> { + let input = fs::canonicalize(input)?; + let output_exists = output.exists(); + let output = if output_exists { + fs::canonicalize(output)? + } else { + let parent = output + .parent() + .filter(|parent| !parent.as_os_str().is_empty()) + .unwrap_or_else(|| Path::new(".")); + let name = output + .file_name() + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "invalid --save path"))?; + fs::canonicalize(parent)?.join(name) + }; + if input == output || (output_exists && files_share_identity(&input, &output)?) { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "--save must not overwrite the --replay input file", + ) + .into()); + } + Ok(()) +} + +#[cfg(unix)] +fn files_share_identity(left: &Path, right: &Path) -> io::Result { + use std::os::unix::fs::MetadataExt; + + let left = fs::metadata(left)?; + let right = fs::metadata(right)?; + Ok(left.dev() == right.dev() && left.ino() == right.ino()) +} + +#[cfg(not(unix))] +fn files_share_identity(_left: &Path, _right: &Path) -> io::Result { + Ok(false) } /// Default on-disk location for the auto-generated CA + server cert. diff --git a/src/session.rs b/src/session.rs new file mode 100644 index 0000000..76e240c --- /dev/null +++ b/src/session.rs @@ -0,0 +1,587 @@ +//! Versioned JSONL persistence for decoded tapgres sessions. +//! +//! Each line is an independent record with a schema version, capture +//! timestamp, record kind, and the structured fields needed by display +//! filters and rich TUI rendering. Replayed records are converted back into +//! [`crate::decode::Output`] so live and file sources share the same renderer. + +use std::collections::VecDeque; +use std::fmt; +use std::fs::File; +use std::io::{self, BufRead, BufReader, LineWriter, Write}; +use std::net::SocketAddr; +use std::path::{Path, PathBuf}; + +use chrono::{Local, SecondsFormat}; +use serde::{Deserialize, Serialize}; + +use crate::decode::{DataColumn, EventDetail, FieldSummary, Output}; +use crate::filter::{DisplayMessage, MessageDirection}; + +/// Current on-disk JSONL schema. Readers refuse every other version so a +/// future incompatible shape cannot be silently misinterpreted. +pub const SCHEMA_VERSION: u32 = 1; + +#[derive(Debug)] +pub enum SessionError { + Io { + path: PathBuf, + source: io::Error, + }, + InvalidRecord { + path: PathBuf, + line: usize, + message: String, + }, + UnsupportedSchema { + path: PathBuf, + line: usize, + found: u32, + }, + Encode { + path: PathBuf, + source: serde_json::Error, + }, +} + +impl fmt::Display for SessionError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Io { path, source } => write!(f, "{}: {source}", path.display()), + Self::InvalidRecord { + path, + line, + message, + } => write!( + f, + "{}: invalid JSONL record at line {line}: {message}", + path.display() + ), + Self::UnsupportedSchema { path, line, found } => write!( + f, + "{}: unsupported schema version {found} at line {line} (supported: {SCHEMA_VERSION})", + path.display() + ), + Self::Encode { path, source } => { + write!( + f, + "{}: could not encode JSONL record: {source}", + path.display() + ) + } + } + } +} + +impl std::error::Error for SessionError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Self::Io { source, .. } => Some(source), + Self::Encode { source, .. } => Some(source), + Self::InvalidRecord { .. } | Self::UnsupportedSchema { .. } => None, + } + } +} + +/// A streaming JSONL writer. It owns the file so a live consumer can record +/// every event before display filtering or in-memory history eviction. +pub struct SessionWriter { + path: PathBuf, + writer: LineWriter, +} + +impl SessionWriter { + pub fn create(path: impl AsRef) -> Result { + let path = path.as_ref().to_path_buf(); + let file = File::create(&path).map_err(|source| SessionError::Io { + path: path.clone(), + source, + })?; + Ok(Self { + path, + // Flush on every completed JSONL line. Recording is opt-in, and + // this prevents the final buffered events from disappearing when + // a long-running CLI capture is stopped with Ctrl-C. + writer: LineWriter::new(file), + }) + } + + pub fn path(&self) -> &Path { + &self.path + } + + pub fn write(&mut self, output: &Output) -> Result<(), SessionError> { + let record = StoredRecord::from_output(output); + serde_json::to_writer(&mut self.writer, &record).map_err(|source| { + SessionError::Encode { + path: self.path.clone(), + source, + } + })?; + self.writer + .write_all(b"\n") + .map_err(|source| SessionError::Io { + path: self.path.clone(), + source, + }) + } + + pub fn flush(&mut self) -> Result<(), SessionError> { + self.writer.flush().map_err(|source| SessionError::Io { + path: self.path.clone(), + source, + }) + } +} + +impl Drop for SessionWriter { + fn drop(&mut self) { + let _ = self.writer.flush(); + } +} + +/// Load a complete saved session. The TUI uses this for `:open`, where the +/// current retained view is replaced atomically only after all lines validate. +pub fn read_all(path: impl AsRef) -> Result, SessionError> { + let mut outputs = Vec::new(); + read_with(path, |output| outputs.push(output))?; + Ok(outputs) +} + +/// Validate a complete session while retaining only its newest `cap` records. +/// The TUI uses this to keep replay memory bounded without exposing a partial +/// file if a later line is malformed. +pub fn read_tail(path: impl AsRef, cap: usize) -> Result<(Vec, usize), SessionError> { + let mut outputs = VecDeque::with_capacity(cap.min(4_096)); + let mut dropped = 0usize; + read_with(path, |output| { + if cap == 0 { + dropped = dropped.saturating_add(1); + return; + } + if outputs.len() == cap { + outputs.pop_front(); + dropped = dropped.saturating_add(1); + } + outputs.push_back(output); + })?; + Ok((outputs.into(), dropped)) +} + +/// Stream a session into a consumer without first retaining the entire file. +/// This is the CLI replay path and keeps large transcripts bounded. +pub fn read_with( + path: impl AsRef, + mut consume: impl FnMut(Output), +) -> Result<(), SessionError> { + let path = path.as_ref().to_path_buf(); + let file = File::open(&path).map_err(|source| SessionError::Io { + path: path.clone(), + source, + })?; + for (index, line) in BufReader::new(file).lines().enumerate() { + let line_number = index + 1; + let line = line.map_err(|source| SessionError::Io { + path: path.clone(), + source, + })?; + if line.trim().is_empty() { + continue; + } + let value: serde_json::Value = + serde_json::from_str(&line).map_err(|source| SessionError::InvalidRecord { + path: path.clone(), + line: line_number, + message: source.to_string(), + })?; + let found = value + .get("schema_version") + .and_then(serde_json::Value::as_u64) + .and_then(|version| u32::try_from(version).ok()) + .ok_or_else(|| SessionError::InvalidRecord { + path: path.clone(), + line: line_number, + message: "schema_version must be an unsigned 32-bit integer".into(), + })?; + if found != SCHEMA_VERSION { + return Err(SessionError::UnsupportedSchema { + path, + line: line_number, + found, + }); + } + let record: StoredRecord = + serde_json::from_value(value).map_err(|source| SessionError::InvalidRecord { + path: path.clone(), + line: line_number, + message: source.to_string(), + })?; + consume(record.into_output(&path, line_number)?); + } + Ok(()) +} + +#[derive(Debug, Serialize, Deserialize)] +struct StoredRecord { + schema_version: u32, + timestamp: String, + #[serde(flatten)] + payload: StoredPayload, +} + +impl StoredRecord { + fn from_output(output: &Output) -> Self { + let (timestamp, payload) = match output { + Output::Message { message, detail } => ( + message.timestamp.clone(), + StoredPayload::Message { + direction: StoredDirection::from(message.direction), + message_type: message.kind.clone(), + text: message.text.clone(), + rendered: message.rendered.clone(), + client: message.client.to_string(), + detail: detail.as_ref().map(StoredDetail::from), + }, + ), + Output::Line(text) => (now_timestamp(), StoredPayload::Line { text: text.clone() }), + Output::Status(text) => ( + now_timestamp(), + StoredPayload::Status { text: text.clone() }, + ), + }; + Self { + schema_version: SCHEMA_VERSION, + timestamp, + payload, + } + } + + fn into_output(self, path: &Path, line: usize) -> Result { + chrono::DateTime::parse_from_rfc3339(&self.timestamp).map_err(|error| { + SessionError::InvalidRecord { + path: path.to_path_buf(), + line, + message: format!("timestamp must be RFC 3339: {error}"), + } + })?; + match self.payload { + StoredPayload::Message { + direction, + message_type, + text, + rendered, + client, + detail, + } => { + let client = + client + .parse::() + .map_err(|error| SessionError::InvalidRecord { + path: path.to_path_buf(), + line, + message: format!("invalid client address {client:?}: {error}"), + })?; + Ok(Output::Message { + message: DisplayMessage { + timestamp: self.timestamp, + rendered, + client, + direction: direction.into(), + kind: message_type, + text, + }, + detail: detail.map(Into::into), + }) + } + StoredPayload::Line { text } => Ok(Output::Line(text)), + StoredPayload::Status { text } => Ok(Output::Status(text)), + } + } +} + +#[derive(Debug, Serialize, Deserialize)] +#[serde(tag = "record_type", rename_all = "snake_case")] +enum StoredPayload { + Message { + direction: StoredDirection, + message_type: String, + text: String, + rendered: String, + client: String, + #[serde(skip_serializing_if = "Option::is_none")] + detail: Option, + }, + Line { + text: String, + }, + Status { + text: String, + }, +} + +#[derive(Clone, Copy, Debug, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +enum StoredDirection { + F2b, + B2f, +} + +impl From for StoredDirection { + fn from(value: MessageDirection) -> Self { + match value { + MessageDirection::FrontendToBackend => Self::F2b, + MessageDirection::BackendToFrontend => Self::B2f, + } + } +} + +impl From for MessageDirection { + fn from(value: StoredDirection) -> Self { + match value { + StoredDirection::F2b => Self::FrontendToBackend, + StoredDirection::B2f => Self::BackendToFrontend, + } + } +} + +#[derive(Debug, Serialize, Deserialize)] +#[serde(tag = "detail_type", rename_all = "snake_case")] +enum StoredDetail { + RowDescription { columns: Vec }, + DataRow { columns: Vec }, +} + +impl From<&EventDetail> for StoredDetail { + fn from(value: &EventDetail) -> Self { + match value { + EventDetail::RowDescription(columns) => Self::RowDescription { + columns: columns.iter().map(StoredFieldSummary::from).collect(), + }, + EventDetail::DataRow(columns) => Self::DataRow { + columns: columns.iter().map(StoredDataColumn::from).collect(), + }, + } + } +} + +impl From for EventDetail { + fn from(value: StoredDetail) -> Self { + match value { + StoredDetail::RowDescription { columns } => { + Self::RowDescription(columns.into_iter().map(Into::into).collect()) + } + StoredDetail::DataRow { columns } => { + Self::DataRow(columns.into_iter().map(Into::into).collect()) + } + } + } +} + +#[derive(Debug, Serialize, Deserialize)] +struct StoredFieldSummary { + name: String, + type_oid: u32, + format_code: i16, +} + +impl From<&FieldSummary> for StoredFieldSummary { + fn from(value: &FieldSummary) -> Self { + Self { + name: value.name.clone(), + type_oid: value.type_oid, + format_code: value.format_code, + } + } +} + +impl From for FieldSummary { + fn from(value: StoredFieldSummary) -> Self { + Self { + name: value.name, + type_oid: value.type_oid, + format_code: value.format_code, + } + } +} + +#[derive(Debug, Serialize, Deserialize)] +struct StoredDataColumn { + name: String, + type_oid: u32, + value: String, +} + +impl From<&DataColumn> for StoredDataColumn { + fn from(value: &DataColumn) -> Self { + Self { + name: value.name.clone(), + type_oid: value.type_oid, + value: value.value.clone(), + } + } +} + +impl From for DataColumn { + fn from(value: StoredDataColumn) -> Self { + Self { + name: value.name, + type_oid: value.type_oid, + value: value.value, + } + } +} + +fn now_timestamp() -> String { + Local::now().to_rfc3339_opts(SecondsFormat::Millis, true) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::fs; + + fn message_output() -> Output { + Output::Message { + message: DisplayMessage { + timestamp: "2026-07-17T12:34:56.789+01:00".into(), + rendered: "[12:34:56.789] [B→F] DataRow: { id='1' }".into(), + client: "127.0.0.1:40005".parse().unwrap(), + direction: MessageDirection::BackendToFrontend, + kind: "DataRow".into(), + text: "{ id='1' }".into(), + }, + detail: Some(EventDetail::DataRow(vec![DataColumn { + name: "id".into(), + type_oid: 23, + value: "'1'".into(), + }])), + } + } + + #[test] + fn jsonl_round_trip_preserves_filter_and_rich_fields() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("capture.jsonl"); + let records = vec![ + Output::Status("capture active".into()), + message_output(), + Output::Line("=== connection closed ===".into()), + ]; + { + let mut writer = SessionWriter::create(&path).unwrap(); + for record in &records { + writer.write(record).unwrap(); + } + writer.flush().unwrap(); + } + + let raw = fs::read_to_string(&path).unwrap(); + assert_eq!(raw.lines().count(), 3); + assert!(raw.contains("\"schema_version\":1")); + assert!(raw.contains("\"record_type\":\"message\"")); + assert!(raw.contains("\"detail_type\":\"data_row\"")); + + let loaded = read_all(&path).unwrap(); + assert_eq!(loaded.len(), 3); + match &loaded[1] { + Output::Message { message, detail } => { + assert_eq!(message.timestamp, "2026-07-17T12:34:56.789+01:00"); + assert_eq!(message.client.port(), 40005); + assert_eq!(message.kind, "DataRow"); + match detail { + Some(EventDetail::DataRow(columns)) => { + assert_eq!(columns[0].name, "id"); + assert_eq!(columns[0].type_oid, 23); + assert_eq!(columns[0].value, "'1'"); + } + other => panic!("expected DataRow detail, got {other:?}"), + } + } + other => panic!("expected message, got {other:?}"), + } + } + + #[test] + fn refuses_unknown_schema_without_partial_tui_load() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("future.jsonl"); + fs::write( + &path, + r#"{"schema_version":99,"timestamp":"now","record_type":"line","text":"future"} +"#, + ) + .unwrap(); + + let error = read_all(&path).unwrap_err().to_string(); + assert!(error.contains("unsupported schema version 99")); + assert!(error.contains("line 1")); + } + + #[test] + fn reports_malformed_json_with_line_number() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("broken.jsonl"); + fs::write( + &path, + concat!( + "{\"schema_version\":1,\"timestamp\":\"2026-07-17T12:34:56.789+01:00\",\"record_type\":\"line\",\"text\":\"valid\"}\n", + "not json\n" + ), + ) + .unwrap(); + + let error = read_all(&path).unwrap_err().to_string(); + assert!(error.contains("line 2")); + } + + #[test] + fn rejects_non_rfc3339_timestamps() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("bad-time.jsonl"); + fs::write( + &path, + r#"{"schema_version":1,"timestamp":"12:34","record_type":"line","text":"bad"} +"#, + ) + .unwrap(); + + let error = read_all(&path).unwrap_err().to_string(); + assert!(error.contains("timestamp must be RFC 3339")); + assert!(error.contains("line 1")); + } + + #[test] + fn tail_reader_validates_all_records_while_bounding_memory() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("many.jsonl"); + let mut writer = SessionWriter::create(&path).unwrap(); + for index in 0..5 { + writer + .write(&Output::Line(format!("line {index}"))) + .unwrap(); + } + writer.flush().unwrap(); + + let (tail, dropped) = read_tail(&path, 2).unwrap(); + assert_eq!(dropped, 3); + assert_eq!(tail.len(), 2); + assert_eq!(tail[0].rendered(), "line 3"); + assert_eq!(tail[1].rendered(), "line 4"); + } + + #[test] + fn reads_the_committed_v1_compatibility_fixture() { + let path = Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/session-v1.jsonl"); + let records = read_all(path).unwrap(); + + assert_eq!(records.len(), 4); + assert!(matches!( + records[2].detail(), + Some(EventDetail::RowDescription(columns)) if columns[0].name == "id" + )); + assert!(matches!( + records[3].detail(), + Some(EventDetail::DataRow(columns)) if columns[0].value == "'1'" + )); + } +} diff --git a/src/tui.rs b/src/tui.rs index c1da222..ca505c0 100644 --- a/src/tui.rs +++ b/src/tui.rs @@ -13,10 +13,12 @@ //! - `r` — toggle rich message rendering //! - `c` — clear //! - `y` — edit the display filter +//! - `/` / `:` — open the command bar (`:save FILE`, `:open FILE`) use crossbeam_channel::Receiver; use std::error::Error; use std::io; +use std::path::PathBuf; use std::sync::Arc; use std::time::Duration; @@ -31,10 +33,14 @@ use crate::capture::PcapOpts; use crate::decode::{self, Output}; use crate::filter::DisplayFilter; use crate::proxy::ProxyOpts; +use crate::session::{self, SessionWriter}; use crate::state::Metrics; /// Cap on retained lines in the TUI's own buffer. const HISTORY_CAP: usize = 50_000; +/// Trim in chunks during fast replay so a producer cannot grow the TUI buffer +/// without bound while avoiding an O(n) front-drain for every single record. +const HISTORY_TRIM_CHUNK: usize = 1_024; /// tapgres ASCII-art banner. Shown by the CLI (`--help` via `before_help`) and /// as the heading of the TUI startup splash. @@ -53,6 +59,7 @@ pub fn run_pcap( metrics: Arc, rich: bool, filter: DisplayFilter, + save: Option, ) -> Result<(), Box> { let splash_lines = pcap_splash_lines(&opts); let source_metrics = metrics.clone(); @@ -67,6 +74,7 @@ pub fn run_pcap( rich, filter, splash_lines, + save, ) } @@ -90,6 +98,7 @@ pub fn run_mitm( metrics: Arc, rich: bool, filter: DisplayFilter, + save: Option, ) -> Result<(), Box> { let splash_lines = mitm_splash_lines(&opts); let source_metrics = metrics.clone(); @@ -114,6 +123,41 @@ pub fn run_mitm( rich, filter, splash_lines, + save, + ) +} + +/// TUI over a saved JSONL session. Records are loaded at full speed through +/// the same channel and rendering path as live capture. +pub fn run_replay( + path: PathBuf, + metrics: Arc, + rich: bool, + filter: DisplayFilter, + save: Option, +) -> Result<(), Box> { + let source_path = path.clone(); + run( + Box::new(move || { + match session::read_tail(&source_path, HISTORY_CAP) { + Ok((outputs, dropped)) => { + if dropped > 0 { + decode::status(format!( + "replay: showing the newest {HISTORY_CAP} records; {dropped} earlier records are outside TUI history" + )); + } + outputs.into_iter().for_each(decode::replay); + } + Err(error) => decode::status(format!("⚠ replay source error: {error}")), + } + decode::close_output(); + }), + "replay", + metrics, + rich, + filter, + Vec::new(), + save, ) } @@ -147,23 +191,26 @@ fn run( rich: bool, filter: DisplayFilter, splash_lines: Vec, + save: Option, ) -> Result<(), Box> { // One channel: the source (background thread) produces via decode::out, // the TUI (this thread) consumes. let (tx, rx) = crossbeam_channel::unbounded(); decode::set_output(tx); + // Validate/create the save destination before starting a live source. + let recorder = save.map(SessionWriter::create).transpose()?; + // The source runs until the process exits; no graceful shutdown here. let _source_thread = std::thread::Builder::new() .name("tapgres-source".into()) .spawn(source)?; let _rate_sampler = metrics.spawn_rate_sampler()?; + let mut app = App::new(rx, mode, metrics, rich, filter, splash_lines); + app.recorder = recorder; let mut terminal = ratatui::try_init()?; - let result = app_loop( - &mut terminal, - App::new(rx, mode, metrics, rich, filter, splash_lines), - ); + let result = app_loop(&mut terminal, app); // Restore the terminal even on error. try_init installs a panic hook that // also restores, so panics are covered too. let _ = ratatui::try_restore(); @@ -194,7 +241,7 @@ struct App { /// `RowDescription` as a typed column list, instead of the flat line. Type /// names are shown with an icon-font (Nerd Font) glyph. rich: bool, - mode: &'static str, + mode: String, metrics: Arc, /// All-time peak messages/sec seen this session, per direction. Used as a /// fixed sparkline scale so bars don't rescale as the rate window slides; @@ -205,11 +252,22 @@ struct App { filter_text: String, filter_error: Option, filter_editing: bool, + command_editing: bool, + command_text: String, + command_notice: Option<(String, bool)>, + /// False after `:open`: the loaded replay replaces the live view for the + /// rest of this TUI session, while the source channel is drained safely. + accept_source_records: bool, + /// Number of events removed by the bounded TUI history before `:save`. + dropped_events: usize, /// Mode-specific connection/capture info lines for the startup splash. splash_lines: Vec, /// Whether the startup splash is still showing. Flips off once a real /// connection is detected (see `app_loop`). show_splash: bool, + /// Optional continuous JSONL recorder. It receives records before the TUI + /// history cap or display filter can hide them. + recorder: Option, } impl App { @@ -232,7 +290,7 @@ impl App { follow: true, wrap: false, rich, - mode, + mode: mode.to_string(), metrics, peak_msgs_in: 0, peak_msgs_out: 0, @@ -240,8 +298,14 @@ impl App { filter_text, filter_error: None, filter_editing: false, + command_editing: false, + command_text: String::new(), + command_notice: None, + accept_source_records: true, + dropped_events: 0, splash_lines, show_splash, + recorder: None, } } @@ -250,11 +314,32 @@ impl App { } fn push_output(&mut self, output: Output) { + let write_error = self + .recorder + .as_mut() + .and_then(|recorder| recorder.write(&output).err()); + if let Some(error) = write_error { + self.recorder = None; + self.command_notice = Some((format!("recording stopped: {error}"), true)); + } let index = self.events.len(); if self.matches(&output) { self.visible.push(index); } self.events.push(output); + if self.events.len() >= HISTORY_CAP + HISTORY_TRIM_CHUNK { + self.trim_history(); + } + } + + fn trim_history(&mut self) { + if self.events.len() <= HISTORY_CAP { + return; + } + let drop_n = self.events.len() - HISTORY_CAP; + self.events.drain(..drop_n); + self.dropped_events = self.dropped_events.saturating_add(drop_n); + self.rebuild_visible(); } fn rebuild_visible(&mut self) { @@ -291,6 +376,88 @@ impl App { self.rebuild_visible(); } + fn execute_command(&mut self) { + let command_text = self.command_text.clone(); + let input = command_text.trim().trim_start_matches([':', '/']).trim(); + let (name, argument) = input + .split_once(char::is_whitespace) + .map(|(name, argument)| (name, argument.trim())) + .unwrap_or((input, "")); + + let result = match name { + "w" | "write" | "save" => self.start_recording(argument), + "o" | "open" => self.open_session(argument), + "" => Err("command is empty".to_string()), + _ => Err(format!("unknown command: {name}")), + }; + self.command_notice = Some(match result { + Ok(message) => (message, false), + Err(message) => (message, true), + }); + self.command_editing = false; + } + + fn start_recording(&mut self, argument: &str) -> Result { + if argument.is_empty() { + return Err("usage: :save FILE".into()); + } + let path = PathBuf::from(argument); + let mut recorder = SessionWriter::create(&path).map_err(|error| error.to_string())?; + for output in &self.events { + recorder.write(output).map_err(|error| error.to_string())?; + } + recorder.flush().map_err(|error| error.to_string())?; + let retained = self.events.len(); + self.recorder = Some(recorder); + let action = if self.accept_source_records { + format!("recording {retained} retained events + future traffic") + } else { + format!("saved {retained} retained replay events") + }; + if self.dropped_events == 0 { + Ok(format!("{action} to {}", path.display())) + } else { + Ok(format!( + "{action} to {}; {} earlier events were outside history", + path.display(), + self.dropped_events + )) + } + } + + fn open_session(&mut self, argument: &str) -> Result { + if argument.is_empty() { + return Err("usage: :open FILE".into()); + } + let path = PathBuf::from(argument); + let (outputs, dropped) = + session::read_tail(&path, HISTORY_CAP).map_err(|error| error.to_string())?; + let count = outputs.len(); + if let Some(recorder) = self.recorder.as_mut() { + recorder.flush().map_err(|error| error.to_string())?; + } + self.recorder = None; + self.events = outputs; + self.visible.clear(); + self.rebuild_visible(); + self.follow = true; + self.mode = "replay".into(); + self.metrics = Arc::new(Metrics::new()); + self.peak_msgs_in = 0; + self.peak_msgs_out = 0; + self.show_splash = false; + self.accept_source_records = false; + self.dropped_events = dropped; + if dropped == 0 { + Ok(format!("opened {count} events from {}", path.display())) + } else { + Ok(format!( + "opened newest {count} events from {}; {dropped} earlier events are outside history", + path.display() + )) + } + } + /// Leave the splash once a real connection has been detected. Startup /// status lines (capture/proxy banner text) arrive before any traffic but /// do not open a connection, so they do not trigger this transition. @@ -304,17 +471,15 @@ impl App { fn app_loop(terminal: &mut ratatui::DefaultTerminal, mut app: App) -> io::Result<()> { loop { while let Ok(record) = app.rx.try_recv() { - app.push_output(record); + if app.accept_source_records { + app.push_output(record); + } } // Leave the splash once a real connection is detected. Startup status // lines (capture/proxy banner text) arrive before any traffic but do // not open a connection, so they don't trigger this transition. app.leave_splash_if_traffic(); - if app.events.len() > HISTORY_CAP { - let drop_n = app.events.len() - HISTORY_CAP; - app.events.drain(..drop_n); - app.rebuild_visible(); - } + app.trim_history(); // 5 (metrics) + 3 (footer) + 2 (log block borders) rows of chrome. let term_h = terminal.size()?.height as usize; @@ -354,6 +519,12 @@ fn app_loop(terminal: &mut ratatui::DefaultTerminal, mut app: App) -> io::Result terminal.draw(|frame| { if app.show_splash { draw_splash(frame, &app); + if app.command_editing { + let [_, command_area] = + Layout::vertical([Constraint::Fill(1), Constraint::Length(3)]) + .areas(frame.area()); + draw_command_bar(frame, &app, command_area); + } } else { draw(frame, &app, log_h); } @@ -385,10 +556,34 @@ fn handle_key(app: &mut App, log_h: usize, key: KeyEvent) -> bool { if key.code == KeyCode::Char('c') && ctrl { return true; } - // On the splash screen only honour quit; everything else is ignored until - // traffic arrives and the main view takes over. + if app.command_editing { + match key.code { + KeyCode::Esc => { + app.command_editing = false; + app.command_text.clear(); + } + KeyCode::Enter => app.execute_command(), + KeyCode::Backspace => { + app.command_text.pop(); + } + KeyCode::Char(ch) if !ctrl => app.command_text.push(ch), + _ => {} + } + return false; + } + // Keep the command bar available before the first connection so a saved + // session can be opened directly from the startup splash. if app.show_splash { - return key.code == KeyCode::Char('q'); + return match key.code { + KeyCode::Char('q') => true, + KeyCode::Char('/') | KeyCode::Char(':') => { + app.command_editing = true; + app.command_text.clear(); + app.command_notice = None; + false + } + _ => false, + }; } if app.filter_editing { match key.code { @@ -440,6 +635,11 @@ fn handle_key(app: &mut App, log_h: usize, key: KeyEvent) -> bool { app.visible.clear(); } KeyCode::Char('y') => app.filter_editing = true, + KeyCode::Char('/') | KeyCode::Char(':') => { + app.command_editing = true; + app.command_text.clear(); + app.command_notice = None; + } KeyCode::Esc if !app.filter.is_empty() => app.clear_filter(), _ => {} } @@ -465,8 +665,18 @@ fn draw_splash(frame: &mut Frame, app: &App) { for line in &app.splash_lines { lines.push(Line::raw(format!(" {line}"))); } + if let Some((notice, is_error)) = &app.command_notice { + lines.push(Line::raw("")); + lines.push(Line::styled( + format!(" {notice}"), + Style::default().fg(if *is_error { Color::Red } else { Color::Green }), + )); + } lines.push(Line::raw("")); - lines.push(Line::styled(" waiting for traffic… press q to quit", dim)); + lines.push(Line::styled( + " waiting for traffic… press / for commands · q to quit", + dim, + )); // Vertically centre the splash block; horizontally centre each line. let height = lines.len() as u16; @@ -670,7 +880,9 @@ fn draw(frame: &mut Frame, app: &App, log_h: usize) { // --- footer: follow/wrap/rich state shown by colour (green = on) --- let on = Style::default().fg(Color::Green); let off = Style::default(); - if app.filter_editing { + if app.command_editing { + draw_command_bar(frame, app, foot_area); + } else if app.filter_editing { let style = if app.filter_error.is_some() { Style::default().fg(Color::Red) } else { @@ -689,8 +901,17 @@ fn draw(frame: &mut Frame, app: &App, log_h: usize) { .block(Block::bordered().title_top(" display filter · Enter done · Esc clear ")), foot_area, ); + } else if let Some((notice, is_error)) = &app.command_notice { + frame.render_widget( + Paragraph::new(Line::styled( + format!(" {notice}"), + Style::default().fg(if *is_error { Color::Red } else { Color::Green }), + )) + .block(Block::bordered().title_top(" status · / command ")), + foot_area, + ); } else { - let footer = Line::from(vec![ + let footer = vec![ Span::raw(" q quit · j/k ↑↓ · PgUp/PgDn · g/G top/bottom · f "), Span::styled("follow", if app.follow { on } else { off }), Span::raw(" · w "), @@ -702,12 +923,27 @@ fn draw(frame: &mut Frame, app: &App, log_h: usize) { "display filter", if app.filter.is_empty() { off } else { on }, ), - Span::raw(" · c clear "), - ]); - frame.render_widget(Paragraph::new(footer).block(Block::bordered()), foot_area); + Span::raw(" · / command · c clear "), + ]; + frame.render_widget( + Paragraph::new(Line::from(footer)).block(Block::bordered()), + foot_area, + ); } } +fn draw_command_bar(frame: &mut Frame, app: &App, area: ratatui::layout::Rect) { + frame.render_widget( + Paragraph::new(Line::from(vec![ + Span::styled(" :", Style::default().fg(Color::Cyan).bold()), + Span::raw(&app.command_text), + Span::styled("█", Style::default().fg(Color::Cyan)), + ])) + .block(Block::bordered().title_top(" command · :save FILE · :open FILE · Esc cancel ")), + area, + ); +} + fn human(value: u64) -> String { const UNITS: [&str; 4] = ["B", "KiB", "MiB", "GiB"]; let mut value = value as f64; @@ -1060,6 +1296,7 @@ mod tests { ) -> Output { Output::Message { message: DisplayMessage { + timestamp: "2026-07-17T12:34:56.789+01:00".into(), rendered: format!("[{kind}] {text}"), client: format!("127.0.0.1:{port}").parse().unwrap(), direction: MessageDirection::FrontendToBackend, @@ -1285,7 +1522,7 @@ mod tests { } #[test] - fn slash_remains_available_for_commands() { + fn slash_opens_command_bar_without_touching_display_filter() { let (_tx, rx) = crossbeam_channel::unbounded(); let mut app = App::new( rx, @@ -1303,6 +1540,75 @@ mod tests { ); assert!(!app.filter_editing); + assert!(app.command_editing); + } + + #[test] + fn save_command_writes_retained_and_future_events() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("saved session.jsonl"); + let mut app = app(); + app.push_output(message("Query", "SELECT 1", 40005)); + app.push_output(message("DataRow", "{ id=1 }", 40005)); + app.filter_text = "message.type == \"Query\"".into(); + app.update_filter(); + + app.command_text = format!("save {}", path.display()); + app.execute_command(); + app.push_output(message("ReadyForQuery", "txn=idle", 40005)); + app.recorder.as_mut().unwrap().flush().unwrap(); + + let saved = session::read_all(&path).unwrap(); + assert_eq!(saved.len(), 3, "display filtering must not affect saving"); + assert!( + app.command_notice + .as_ref() + .is_some_and(|(message, error)| !error && message.contains("recording")) + ); + } + + #[test] + fn open_command_atomically_replaces_view_and_preserves_rich_detail() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("replay.jsonl"); + let replayed = message_with_detail("DataRow", "{ id=1 }", 40005, Some(data_row_detail(2))); + let mut writer = SessionWriter::create(&path).unwrap(); + writer.write(&replayed).unwrap(); + writer.flush().unwrap(); + + let mut app = app(); + app.push_output(message("Query", "old", 40005)); + app.command_text = format!("open {}", path.display()); + app.execute_command(); + + assert_eq!(app.mode, "replay"); + assert!(!app.accept_source_records); + assert_eq!(app.events.len(), 1); + assert!(matches!( + app.events[0].detail(), + Some(EventDetail::DataRow(columns)) if columns.len() == 2 + )); + } + + #[test] + fn failed_open_keeps_the_existing_view() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("broken.jsonl"); + std::fs::write(&path, "not json\n").unwrap(); + let mut app = app(); + app.push_output(message("Query", "SELECT 1", 40005)); + + app.command_text = format!("open {}", path.display()); + app.execute_command(); + + assert_eq!(app.mode, "test"); + assert!(app.accept_source_records); + assert_eq!(app.events.len(), 1); + assert!( + app.command_notice + .as_ref() + .is_some_and(|(message, error)| *error && message.contains("invalid JSONL")) + ); } #[test] @@ -1364,7 +1670,7 @@ mod tests { } #[test] - fn splash_only_honours_quit() { + fn splash_honours_quit_and_command_bar() { let (_tx, rx) = crossbeam_channel::unbounded(); let mut app = App::new( rx, @@ -1381,6 +1687,15 @@ mod tests { KeyEvent::new(KeyCode::Char('y'), KeyModifiers::NONE), )); assert!(!app.filter_editing); + // Commands remain available so a replay can be opened before live + // traffic arrives. + assert!(!handle_key( + &mut app, + 10, + KeyEvent::new(KeyCode::Char('/'), KeyModifiers::NONE), + )); + assert!(app.command_editing); + app.command_editing = false; // `q` quits even from the splash. assert!(handle_key( &mut app, diff --git a/tests/fixtures/session-v1.jsonl b/tests/fixtures/session-v1.jsonl new file mode 100644 index 0000000..97e8352 --- /dev/null +++ b/tests/fixtures/session-v1.jsonl @@ -0,0 +1,4 @@ +{"schema_version":1,"timestamp":"2026-07-17T12:34:56.700+01:00","record_type":"status","text":"tapgres fixture session"} +{"schema_version":1,"timestamp":"2026-07-17T12:34:56.789+01:00","record_type":"message","direction":"f2b","message_type":"Query","text":"SELECT id FROM orders","rendered":"[12:34:56.789] [F→B] Query: SELECT id FROM orders","client":"127.0.0.1:40005"} +{"schema_version":1,"timestamp":"2026-07-17T12:34:56.800+01:00","record_type":"message","direction":"b2f","message_type":"RowDescription","text":"id(oid=23, text)","rendered":"[12:34:56.800] [B→F] RowDescription: id(oid=23, text)","client":"127.0.0.1:40005","detail":{"detail_type":"row_description","columns":[{"name":"id","type_oid":23,"format_code":0}]}} +{"schema_version":1,"timestamp":"2026-07-17T12:34:56.810+01:00","record_type":"message","direction":"b2f","message_type":"DataRow","text":"{ id='1' }","rendered":"[12:34:56.810] [B→F] DataRow: { id='1' }","client":"127.0.0.1:40005","detail":{"detail_type":"data_row","columns":[{"name":"id","type_oid":23,"value":"'1'"}]}} diff --git a/tests/session_cli.rs b/tests/session_cli.rs new file mode 100644 index 0000000..92916b4 --- /dev/null +++ b/tests/session_cli.rs @@ -0,0 +1,185 @@ +//! CLI integration coverage for the durable JSONL session source. + +use std::process::Command; + +use tapgres::decode::{EventDetail, FieldSummary, Output}; +use tapgres::filter::{DisplayMessage, MessageDirection}; +use tapgres::session::{self, SessionWriter}; + +fn message(kind: &str, text: &str, direction: MessageDirection) -> Output { + let tag = match direction { + MessageDirection::FrontendToBackend => "F→B", + MessageDirection::BackendToFrontend => "B→F", + }; + Output::Message { + message: DisplayMessage { + timestamp: "2026-07-17T12:34:56.789+01:00".into(), + rendered: format!("[12:34:56.789] [{tag}] {kind}: {text}"), + client: "127.0.0.1:40005".parse().unwrap(), + direction, + kind: kind.into(), + text: text.into(), + }, + detail: (kind == "RowDescription").then(|| { + EventDetail::RowDescription(vec![FieldSummary { + name: "id".into(), + type_oid: 23, + format_code: 0, + }]) + }), + } +} + +fn write_fixture(path: &std::path::Path) { + let mut writer = SessionWriter::create(path).unwrap(); + writer + .write(&Output::Status("saved session".into())) + .unwrap(); + writer + .write(&message( + "Query", + "SELECT * FROM orders", + MessageDirection::FrontendToBackend, + )) + .unwrap(); + writer + .write(&message( + "RowDescription", + "id(oid=23, text)", + MessageDirection::BackendToFrontend, + )) + .unwrap(); + writer.flush().unwrap(); +} + +#[test] +fn replay_uses_the_normal_stdout_filter_path() { + let dir = tempfile::tempdir().unwrap(); + let input = dir.path().join("capture.jsonl"); + write_fixture(&input); + + let result = Command::new(env!("CARGO_BIN_EXE_tapgres")) + .args([ + "--replay", + input.to_str().unwrap(), + "--display-filter", + "message.type == \"Query\"", + ]) + .output() + .unwrap(); + + assert!( + result.status.success(), + "stderr: {}", + String::from_utf8_lossy(&result.stderr) + ); + let stdout = String::from_utf8(result.stdout).unwrap(); + let stderr = String::from_utf8(result.stderr).unwrap(); + assert!(stdout.contains("Query: SELECT * FROM orders")); + assert!(!stdout.contains("RowDescription")); + assert!(stderr.contains("saved session")); +} + +#[test] +fn save_records_unfiltered_replay_stream() { + let dir = tempfile::tempdir().unwrap(); + let input = dir.path().join("capture.jsonl"); + let output = dir.path().join("copy.jsonl"); + write_fixture(&input); + + let result = Command::new(env!("CARGO_BIN_EXE_tapgres")) + .args([ + "--replay", + input.to_str().unwrap(), + "--save", + output.to_str().unwrap(), + "-Y", + "message.type == \"Query\"", + ]) + .output() + .unwrap(); + + assert!( + result.status.success(), + "stderr: {}", + String::from_utf8_lossy(&result.stderr) + ); + let saved = session::read_all(&output).unwrap(); + assert_eq!(saved.len(), 3); + assert!(saved.iter().any(|record| matches!( + record, + Output::Message { + message, + detail: Some(EventDetail::RowDescription(columns)), + } if message.kind == "RowDescription" && columns[0].type_oid == 23 + ))); +} + +#[test] +fn replay_refuses_to_overwrite_its_input() { + let dir = tempfile::tempdir().unwrap(); + let input = dir.path().join("capture.jsonl"); + write_fixture(&input); + + let result = Command::new(env!("CARGO_BIN_EXE_tapgres")) + .args([ + "--replay", + input.to_str().unwrap(), + "--save", + input.to_str().unwrap(), + ]) + .output() + .unwrap(); + + assert!(!result.status.success()); + assert!(String::from_utf8_lossy(&result.stderr).contains("must not overwrite")); + assert_eq!(session::read_all(&input).unwrap().len(), 3); +} + +#[test] +fn replay_and_save_accept_relative_paths() { + let dir = tempfile::tempdir().unwrap(); + write_fixture(&dir.path().join("capture.jsonl")); + + let result = Command::new(env!("CARGO_BIN_EXE_tapgres")) + .current_dir(dir.path()) + .args(["--replay", "capture.jsonl", "--save", "copy.jsonl"]) + .output() + .unwrap(); + + assert!( + result.status.success(), + "stderr: {}", + String::from_utf8_lossy(&result.stderr) + ); + assert_eq!( + session::read_all(dir.path().join("copy.jsonl")) + .unwrap() + .len(), + 3 + ); +} + +#[cfg(unix)] +#[test] +fn replay_refuses_hard_link_alias_as_save_target() { + let dir = tempfile::tempdir().unwrap(); + let input = dir.path().join("capture.jsonl"); + let alias = dir.path().join("same-file.jsonl"); + write_fixture(&input); + std::fs::hard_link(&input, &alias).unwrap(); + + let result = Command::new(env!("CARGO_BIN_EXE_tapgres")) + .args([ + "--replay", + input.to_str().unwrap(), + "--save", + alias.to_str().unwrap(), + ]) + .output() + .unwrap(); + + assert!(!result.status.success()); + assert!(String::from_utf8_lossy(&result.stderr).contains("must not overwrite")); + assert_eq!(session::read_all(&input).unwrap().len(), 3); +}