diff --git a/CHANGELOG.md b/CHANGELOG.md index acafde0..de2c602 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,6 +22,7 @@ User-visible changes to `voltage`, newest first. Each pull request that changes - `voltage wallets get` on a terminal leads with the wallet's name, its balances (with a USD estimate on mainnet), and a 14-day sparkline of completed payments. - `--qr` on a terminal draws the invoice black on white with a quiet zone, so it scans on dark themes, under a `Scan to pay` caption with the amount, memo, and expiry, and after the result so it stays on screen; `--wait completed` counts down to the expiry. The caption leads with the network, its dot in the website's color. - A payment that settles on a terminal closes with `⚡ Payment ID completed` instead of a check mark, printed after the payment's record so it stays on screen. +- `voltage payments list --watch` follows payments until Ctrl-C: a live table that marks new and changed payments on a terminal, or one `snapshot`, `new`, or `updated` event per payment when piped or with `--json`. `--interval` sets the refresh (default 5 seconds). ## 0.2.0 - 2026-10-22 diff --git a/docs/guides/payments.md b/docs/guides/payments.md index 8f8b5c1..d69eb07 100644 --- a/docs/guides/payments.md +++ b/docs/guides/payments.md @@ -36,6 +36,17 @@ An accepted submission returns immediately with `outcome: accepted`. In human ou voltage payments get PAYMENT_ID --profile prod --wait completed --timeout 120 ``` +## Watch payments + +`payments list --watch` keeps reading the list every `--interval` seconds (default 5) until you press Ctrl-C, with the same filters as `payments list`: + +```sh +voltage payments list --profile prod --wallet WALLET_ID --watch +voltage payments list --profile prod --watch --interval 2 --json +``` + +On a terminal it is a live table on the alternate screen, newest first, with `●` beside each payment that is new or changed since the last refresh; Ctrl-C puts your screen back and prints `Stopped watching payments.` (exit code 130). Piped or with `--json`, it writes one event envelope per payment instead: `snapshot` for each payment in the first read, then `new` or `updated` as they appear, with the payment in `data` and its ID in `id`. Each refresh reads the first page of the most recently updated payments (`sort_key=updated_at`, `sort_order=DESC`), so `--watch` cannot be combined with `--all`, `--cursor`, `--sort-key`, or `--sort-order`; more changes than one page holds (`--limit`, the API's default otherwise) between two refreshes are not all reported. A refresh that fails in transport, with a server error (5xx), or with a rate limit (429) is retried after the server's `Retry-After`, or after a pause that grows up to a minute, and the live view says why. A rejected credential (401) is resolved again once, so a saved login exchanges a fresh organization token. Any other error ends the watch. `--timeout` limits each refresh's request, not the watch, and `--quiet` also hides `Stopped watching payments.` On a terminal, rows beyond its height are left out and counted. + ## Show an invoice For a BOLT11 receive, `--qr` prints the invoice as a terminal QR code and `--copy` puts it on the clipboard as soon as it is ready. Either flag implies `--wait ready`; with `--wait completed` the invoice is shown first and polling continues to settlement. On a terminal with color the QR code is black on white with the QR spec's four-module quiet zone, so it scans on a dark theme too; with color off (`NO_COLOR`, `--color never`, `TERM=dumb`) it is the compact code, which follows your terminal's theme and may read inverted on a dark one. It appears under a bold `Scan to pay` caption with the amount, memo, and expiry when the payment has them. The caption starts with the network, its dot in the website's color (mainnet orange, mutinynet and signet purple, testnet green); the code itself stays black and white. The CLI reads the payment's wallet for its network, since mutinynet and signet invoices share a prefix, and uses the prefix if that read fails. A wait that ends at `ready` shows the code after the result, so the code stays on screen; with `--wait completed` it comes first, and the status line counts down to the expiry. Redirected stderr gets the compact code it always did. diff --git a/src/api.rs b/src/api.rs index 1023187..148f22b 100644 --- a/src/api.rs +++ b/src/api.rs @@ -18,15 +18,16 @@ use crate::{ dry_run, error::{CredentialSources, ErrorDetail, ErrorKind, ResponseLimit}, input::{PaginationMode, Request}, - output::{Envelope, Outcome, Output}, + output::{Envelope, LiveScreen, Outcome, Output}, payment::{Currency, PaymentDirection, PaymentView, ReceiveKind, WaitProgress, WaitTarget}, price::{PRICE_URL, PriceService, UsdPerBtc}, qr, registry::{AuthScheme, Method, Operation, OperationId, operation}, secret::Secret, - style::Depth, - terminal::Terminal, + style::{Depth, Line, Role}, + terminal::{self, Terminal}, timeline::Timeline, + watch, }; use futures_util::StreamExt; use serde::{Deserialize, Deserializer, Serialize}; @@ -35,7 +36,10 @@ use std::{ borrow::Cow, collections::BTreeSet, io::Write, - sync::{Mutex, OnceLock, PoisonError}, + sync::{ + Mutex, OnceLock, PoisonError, + atomic::{AtomicBool, Ordering}, + }, time::{Duration, SystemTime, UNIX_EPOCH}, }; use tokio::time::Instant; @@ -290,6 +294,8 @@ impl Api { ErrorKind::for_http_status(status) }; Error { + response_status: Some(status), + retry_after, hint: self.authorization.rejection_hint(status), // Only the sources are attached: an HTTP detail would make an HTML 404 // read as a missing resource. @@ -322,6 +328,7 @@ impl Api { if let Some(hint) = hint.or_else(|| rejection.hint().map(Cow::Borrowed)) { error = error.with_hint(hint); } + error.retry_after = retry_after; return Err(error .with_detail(ErrorDetail::Http { http_status: status, @@ -635,6 +642,7 @@ pub async fn execute( let call = Call { api, global, + resolver, invocation, scope, deadline: Instant::now() + global.timeout, @@ -643,7 +651,12 @@ pub async fn execute( }; match run { Run::EventStream => call.api.stream(&call.request, call.out).await, - Run::Read(read) => call.read(read).await, + Run::Read(read) => { + if matches!(read.then, ReadThen::Watch { .. }) { + resolver.submission.watching(); + } + call.read(read).await + } Run::Mutation(mutation) => { call.mutate(mutation, resolver.settings, resolver.submission) .await @@ -804,6 +817,8 @@ pub struct Read { pub enum ReadThen { /// Write the response, then follow further pages when `follow` (`--all`) is set. Pages { follow: bool }, + /// `payments list --watch`: read again every `interval_seconds` until interrupted. + Watch { interval_seconds: u64 }, /// A checkout session read answers 202 until its projection exists. SessionProjection, /// Poll the payment until `until`. A submitted payment can be missing from the read @@ -825,6 +840,9 @@ impl Read { && invocation.wait.is_some(), }, None if operation.id == OperationId::GetSession => ReadThen::SessionProjection, + None if let Some(interval) = invocation.watch => ReadThen::Watch { + interval_seconds: interval.as_secs(), + }, None => ReadThen::Pages { follow: invocation.all, }, @@ -868,6 +886,8 @@ impl Mutation { struct Call<'a> { api: Api, global: &'a GlobalFlags, + /// How the credential was chosen, to choose it again when a watch outlives it. + resolver: Resolver<'a>, invocation: &'a ApiInvocation, scope: &'a Scope, deadline: Instant, @@ -878,6 +898,9 @@ struct Call<'a> { impl Call<'_> { async fn read(mut self, read: Read) -> Result<()> { let operation = self.invocation.operation; + if matches!(read.then, ReadThen::Watch { .. }) { + watch::follow_updates(&mut self.request.query); + } let sent = self .api .send( @@ -911,6 +934,10 @@ impl Call<'_> { let response = self.await_session_projection(response).await?; self.write_pages(response, Outcome::Retrieved, false).await } + ReadThen::Watch { interval_seconds } => { + self.watch(response, Duration::from_secs(interval_seconds)) + .await + } ReadThen::Pages { follow } => { if operation.id == OperationId::GetWallet && self.out.on_terminal() { self.lead_with_balance(&response.body).await?; @@ -920,6 +947,74 @@ impl Call<'_> { } } + /// `payments list --watch`: read the list again every `interval` until interrupted, most + /// recently updated first. An interactive stdout gets a live table on the alternate + /// screen that marks payments new or changed since the last read; other output gets one + /// event envelope per payment, `snapshot` for the first read, then `new` or `updated`. + /// A refresh the server could not serve (a 5xx, a 429, or a transport failure) is retried + /// after its `Retry-After`, or after a growing pause; a rejected credential (401) is + /// resolved again once, since a saved login's organization token is short-lived; any + /// other failure ends the watch. + async fn watch(&mut self, mut response: Response, interval: Duration) -> Result<()> { + // Each read would otherwise draw a status line over the table. + self.api.terminal = self.api.terminal.quieted(); + let live = self.out.on_terminal(); + let _screen = live.then(LiveScreen::enter); + let mut seen = watch::Seen::default(); + let mut failure = None; + let mut backoff = Backoff::new(interval, watch::BACKOFF_CEILING); + let mut pause = response.retry_after.unwrap_or(interval); + let mut reauthorized = false; + loop { + let changes = seen.update(&response.body); + if live { + let marked: Vec = changes + .iter() + .filter(|(_, _, change)| *change != watch::Change::Snapshot) + .map(|(id, _, _)| id.clone()) + .collect(); + let header = watch_header(interval, failure.as_ref(), self.api.terminal); + self.out.live_frame(&header, &response.body, &marked)?; + } else { + for (id, payment, change) in changes { + let event = Envelope::event(payment, change.as_str().into(), Some(id)); + self.out.write(event, &self.api.secrets())?; + } + } + tokio::time::sleep(pause).await; + let operation = self.invocation.operation; + let sent = self + .api + .send( + operation.method, + &self.request.path, + &self.request.query, + None, + ) + .await; + match sent { + Ok(next) => { + pause = next.retry_after.unwrap_or(interval); + response = next; + failure = None; + reauthorized = false; + backoff = Backoff::new(interval, watch::BACKOFF_CEILING); + } + Err(error) if error.response_status() == Some(401) && !reauthorized => { + self.api.authorization = + Authorization::resolve(self.invocation, self.resolver, self.scope).await?; + reauthorized = true; + pause = Duration::ZERO; + } + Err(error) if watch::retries(&error) => { + pause = error.retry_after.unwrap_or_else(|| backoff.pause()); + failure = Some(error); + } + Err(error) => return Err(error), + } + } + } + /// On a terminal, put the wallet's balance and recent activity above its fields. The /// price and payments reads are best effort: a failure leaves its line out. async fn lead_with_balance(&self, wallet: &Value) -> Result<()> { @@ -1233,6 +1328,8 @@ pub struct SubmissionState { /// The saved login whose profile offer is under way. Kept apart from `session`: the /// offer's own refresh or token exchange sets and settles a session change of its own. profile_offer: Mutex>, + /// `payments list --watch` is running, so Ctrl-C is how it ends rather than a cut-off. + watching: AtomicBool, } /// A login change that the server and the saved credentials can disagree about. @@ -1273,6 +1370,16 @@ impl SubmissionState { *self.session.lock().unwrap_or_else(PoisonError::into_inner) = None; } + /// Marks a watch, which only reads and ends with Ctrl-C. + fn watching(&self) { + self.watching.store(true, Ordering::Relaxed); + } + + /// A watch was running, so Ctrl-C is how it ended rather than a failure to report. + pub fn is_watching(&self) -> bool { + self.watching.load(Ordering::Relaxed) + } + /// Marks the profile offer after a login, which is saved by then, as under way or done. pub(crate) fn offering_profile(&self, account: Option<&str>) { *self @@ -1506,6 +1613,25 @@ impl Call<'_> { } } +/// The live watch's first line: what it is, how often it refreshes, how to stop, and why +/// the last refresh failed, if it did. +fn watch_header(interval: Duration, failure: Option<&Error>, terminal: Terminal) -> Line { + let separator = terminal.glyphs().separator; + let mut line = Line::default() + .styled(Role::Emphasis, "Watching payments") + .text(&format!( + " {separator} every {} {separator} Ctrl-C to stop", + terminal::elapsed(interval) + )); + if let Some(error) = failure { + line = line + .text(&format!(" {separator} ")) + .paint(Role::Caution, "last refresh failed:") + .text(&format!(" {}", error.message)); + } + line +} + /// Copy and render a ready BOLT11 invoice as requested; both go to stderr. fn present_invoice( invoice: &str, @@ -1696,6 +1822,29 @@ fn next_page(body: &Value, request: &Request) -> Result #[cfg(test)] mod tests { use super::*; + + #[test] + fn the_watch_header_says_how_to_stop_and_why_a_refresh_failed() { + let terminal = Terminal::new(false, true); + let separator = terminal.glyphs().separator; + let shown = |line: Line| anstream::adapter::strip_str(line.as_str()).to_string(); + assert_eq!( + shown(watch_header(Duration::from_secs(5), None, terminal)), + format!("Watching payments {separator} every 5s {separator} Ctrl-C to stop") + ); + let failure = Error::transport("Connection reset"); + assert_eq!( + shown(watch_header( + Duration::from_secs(5), + Some(&failure), + terminal + )), + format!( + "Watching payments {separator} every 5s {separator} Ctrl-C to stop \ + {separator} last refresh failed: Connection reset" + ) + ); + } use crate::{ cli::{Command, Invocation}, config::Config, diff --git a/src/cli.rs b/src/cli.rs index 4ac719d..6b5422f 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -399,6 +399,28 @@ struct WaitFlags { wait: Option, } +#[derive(Debug, Args)] +struct WatchFlags { + /// Keep listing until Ctrl-C: a live table on a terminal, else one event per new or + /// changed payment + #[arg( + long, + conflicts_with_all = ["all", "cursor", "sort-key", "sort-order"], + help_heading = "Waiting and presentation" + )] + watch: bool, + /// Seconds between refreshes while watching + #[arg( + long, + value_name = "SECONDS", + default_value = "5", + value_parser = parse_timeout, + requires = "watch", + help_heading = "Waiting and presentation" + )] + interval: Duration, +} + #[derive(Debug, Args)] struct InvoiceFlags { /// Render a ready BOLT11 invoice as a QR code to scan; implies --wait ready @@ -622,6 +644,8 @@ pub struct ApiInvocation { pub token_file: Option, pub origin: Option, pub wait: Option, + /// `payments list --watch`: refresh at this interval until interrupted. + pub watch: Option, pub qr: bool, pub copy: bool, pub body: Option, @@ -812,6 +836,7 @@ fn api_invocation( let checkout = parse_if::(operation.auth.is_checkout(), m)?; let origin = parse_if::(takes_origin(operation), m)?; let wait = parse_if::(operation.id.supports_wait(), m)?; + let watch = parse_if::(operation.id == OperationId::GetPayments, m)?; let invoice = parse_if::(operation.id.presents_invoice(), m)?; let data = parse_if::(operation.body, m)?; let body = match data { @@ -832,6 +857,9 @@ fn api_invocation( token_file: checkout.and_then(|flags| flags.token_file), origin: origin.and_then(|flags| flags.origin), wait: wait.and_then(|flags| flags.wait), + watch: watch + .filter(|flags| flags.watch) + .map(|flags| flags.interval), qr: invoice.as_ref().is_some_and(|flags| flags.qr), copy: invoice.as_ref().is_some_and(|flags| flags.copy), body, @@ -1056,6 +1084,9 @@ fn endpoint(mut cmd: ClapCommand, operation: &Operation, about: String) -> ClapC if takes_origin(operation) { cmd = OriginFlags::augment_args(cmd); } + if operation.id == OperationId::GetPayments { + cmd = WatchFlags::augment_args(cmd); + } if operation.id.supports_wait() { cmd = WaitFlags::augment_args(cmd); } @@ -1587,6 +1618,40 @@ mod tests { assert!(api.body.is_none()); } + #[test] + fn only_payments_list_watches_and_its_interval_needs_the_watch() { + let watched = api(&["payments", "list", "--watch"]); + assert_eq!(watched.watch, Some(Duration::from_secs(5))); + let quick = api(&["payments", "list", "--watch", "--interval", "2"]); + assert_eq!(quick.watch, Some(Duration::from_secs(2))); + assert_eq!(api(&["payments", "list"]).watch, None); + for args in [ + &["voltage", "payments", "list", "--interval", "2"][..], + &["voltage", "payments", "list", "--watch", "--all"], + &["voltage", "payments", "list", "--watch", "--cursor", "c1"], + &[ + "voltage", + "payments", + "list", + "--watch", + "--sort-key", + "created_at", + ], + &[ + "voltage", + "payments", + "list", + "--watch", + "--sort-order", + "ASC", + ], + &["voltage", "payments", "list", "--watch", "--interval", "0"], + &["voltage", "wallets", "list", "--watch"], + ] { + assert!(Invocation::try_parse_from(args).is_err(), "{args:?}"); + } + } + #[test] fn payment_aliases_reach_create_payment_with_their_direction() { let invocation = api(&[ diff --git a/src/error.rs b/src/error.rs index e0e14a6..8114aef 100644 --- a/src/error.rs +++ b/src/error.rs @@ -162,6 +162,11 @@ pub struct Error { pub hint: Option>, pub detail: Option, pub(crate) stdout_closed: bool, + /// The HTTP status of a response that came back as an error, also when the response was + /// not JSON and so has no HTTP detail. Only retry decisions read it. + pub(crate) response_status: Option, + /// The server's `Retry-After` on that response. + pub(crate) retry_after: Option, } /// How long a request had to get a response. @@ -303,6 +308,8 @@ impl Error { hint: None, detail: None, stdout_closed: false, + response_status: None, + retry_after: None, } } @@ -357,6 +364,11 @@ impl Error { } } + /// The status of the response this error came from, whether or not it carried JSON. + pub fn response_status(&self) -> Option { + self.http_status().or(self.response_status) + } + pub fn is_transport(&self) -> bool { self.kind == ErrorKind::Transport } diff --git a/src/lib.rs b/src/lib.rs index ea1d6be..bed9218 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -28,6 +28,7 @@ mod terminal; mod timeline; mod wallet_currency; mod wallet_shortlist; +mod watch; mod wordmark; pub(crate) use error::{Error, Result}; diff --git a/src/output.rs b/src/output.rs index e5fbbb7..89f2d3d 100644 --- a/src/output.rs +++ b/src/output.rs @@ -262,6 +262,57 @@ impl Output { write_stdout_styled(text.as_bytes(), self.color) } + /// One frame of a live watch on an interactive stdout: clear the screen, then `header` + /// and the payments table, with a marker beside each row whose ID is in `marked`. + pub fn live_frame(&self, header: &Line, page: &Value, marked: &[String]) -> Result<()> { + let view = human::View { + width: terminal::width().map(|width| width.saturating_sub(MARK_WIDTH)), + color: true, + glyphs: Glyphs::detect(|name| std::env::var(name).ok()), + ..human::View::plain() + }; + // Rows past the terminal's height would scroll the header off; the newest updates + // come first, so the rows kept are the ones that matter. + let rows = terminal::height().map(|rows| rows.saturating_sub(FRAME_LINES)); + let (page, more) = crate::watch::first(page, rows); + let mut table = Vec::new(); + let envelope = Envelope::new(Some(200), page.clone(), None, Outcome::Retrieved); + human::render(&mut table, &envelope, view)?; + let ids: Vec> = crate::watch::items(&page) + .iter() + .map(|payment| payment.get("id").and_then(Value::as_str)) + .collect(); + let mark = Line::default().paint( + Role::Pending, + if view.glyphs == Glyphs::UNICODE { + "●" + } else { + "*" + }, + ); + let mut frame = format!("{}\n\n", header.as_str()); + // A next cursor is for paging by hand, which a live view does not do. + let table = String::from_utf8_lossy(&table); + let lines = table + .lines() + .filter(|line| !line.starts_with("Next cursor:")); + for (index, line) in lines.enumerate() { + // The table's first line is its header; row `n` is the `n`th payment. + let marked = index + .checked_sub(1) + .and_then(|row| ids.get(row).copied().flatten()) + .is_some_and(|id| marked.iter().any(|marked| marked == id)); + let prefix = if marked { mark.as_str() } else { " " }; + frame.push_str(&format!("{prefix} {line}\n")); + } + if more > 0 { + frame.push_str(&format!(" … {more} more, below the terminal's height\n")); + } + // Home and clear are screen control, not color, so they skip the color stream. + finish_stdout(std::io::stdout().lock(), b"\x1b[H\x1b[2J")?; + write_stdout_styled(frame.as_bytes(), self.color) + } + /// JSON output collects `--all` pages into one envelope; NDJSON streams them. pub fn collects_pages(&self) -> bool { self.format == OutputFormat::Json @@ -298,6 +349,30 @@ impl Output { } } +/// Columns the live watch's row marker takes: the mark and a space. +const MARK_WIDTH: usize = 2; +/// Lines a live frame needs besides its rows: header, blank, table header, hidden columns, +/// result count, the `more` line, and the line the cursor rests on. +const FRAME_LINES: usize = 7; + +/// The terminal's alternate screen with the cursor hidden, for a live watch. Dropping it +/// puts back the person's screen and cursor, including when Ctrl-C ends the watch. +#[must_use = "the alternate screen is left as soon as the guard is dropped"] +pub struct LiveScreen; + +impl LiveScreen { + pub fn enter() -> Self { + let _ = finish_stdout(std::io::stdout().lock(), b"\x1b[?1049h\x1b[?25l"); + Self + } +} + +impl Drop for LiveScreen { + fn drop(&mut self) { + let _ = finish_stdout(std::io::stdout().lock(), b"\x1b[?25h\x1b[?1049l"); + } +} + /// Write rendered output; a closed stdout ends the pipeline successfully. pub fn write_stdout(bytes: &[u8]) -> Result<()> { finish_stdout(std::io::stdout().lock(), bytes) diff --git a/src/startup.rs b/src/startup.rs index 50b0cb9..e742298 100644 --- a/src/startup.rs +++ b/src/startup.rs @@ -10,7 +10,7 @@ use crate::{ }, config::{self, Profile, Scope, Settings}, context, - error::CredentialSources, + error::{CredentialSources, ErrorKind}, output::{Envelope, Output, OutputFormat, report_error, write_stdout}, pick, price::{PRICE_URL, PriceService}, @@ -33,6 +33,7 @@ pub async fn run() -> ExitCode { }; let format = invocation.global.output_format(); let color = invocation.global.color(); + let quiet = invocation.global.quiet; let submission = SubmissionState::default(); // Biased, with the interrupt first: tokio installs the SIGINT handler when `ctrl_c` is // first polled. Polled second, a command that reaches a prompt without awaiting would leave @@ -56,6 +57,13 @@ pub async fn run() -> ExitCode { match result { Ok(()) => ExitCode::SUCCESS, Err(error) if error.is_stdout_closed() => ExitCode::SUCCESS, + // Ctrl-C is how a watch ends; it keeps the interrupt's exit code without an error. + Err(error) if error.kind == ErrorKind::Interrupted && submission.is_watching() => { + if !quiet { + eprintln!("Stopped watching payments."); + } + process_exit(error.kind.exit_code()) + } Err(error) => fail(error, format, color), } } diff --git a/src/terminal.rs b/src/terminal.rs index 5ba260e..82fb568 100644 --- a/src/terminal.rs +++ b/src/terminal.rs @@ -268,6 +268,17 @@ pub fn width() -> Option { width_when(std::io::stdout().is_terminal()) } +/// The terminal's height in rows while stdout is interactive, for a screen that redraws. +pub fn height() -> Option { + if !std::io::stdout().is_terminal() { + return None; + } + crossterm::terminal::size() + .ok() + .map(|(_, rows)| usize::from(rows)) + .filter(|rows| *rows > 0) +} + /// `width` for stderr, where status lines go. pub fn stderr_width() -> Option { width_when(std::io::stderr().is_terminal()) diff --git a/src/watch.rs b/src/watch.rs new file mode 100644 index 0000000..4b38719 --- /dev/null +++ b/src/watch.rs @@ -0,0 +1,189 @@ +//! What changed between two reads of a payments list, for `payments list --watch`. + +use crate::Error; +use serde_json::Value; +use std::{collections::HashMap, time::Duration}; + +/// The longest pause between refreshes after repeated failures. +pub const BACKOFF_CEILING: Duration = Duration::from_secs(60); + +/// A watch compares pages of the most recently updated payments, so a change always lands on +/// the first page; a cursor or another order would hide it. +pub fn follow_updates(query: &mut Vec<(String, String)>) { + query.retain(|(name, _)| !matches!(name.as_str(), "sort_key" | "sort_order" | "cursor")); + query.extend( + [("sort_key", "updated_at"), ("sort_order", "DESC")] + .map(|(name, value)| (name.to_owned(), value.to_owned())), + ); +} + +/// A refresh the server could not serve now: a transport failure, a 429, or any 5xx, with or +/// without a JSON body. +pub fn retries(error: &Error) -> bool { + error.is_transport() || matches!(error.response_status(), Some(429 | 500..=599)) +} + +/// How a payment appears in a watch: in the first read, new since the last read, or +/// changed since then. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum Change { + Snapshot, + New, + Updated, +} + +impl Change { + pub fn as_str(self) -> &'static str { + match self { + Self::Snapshot => "snapshot", + Self::New => "new", + Self::Updated => "updated", + } + } +} + +/// Each payment's last seen state, by ID. +#[derive(Debug, Default)] +pub struct Seen { + states: HashMap, + started: bool, +} + +impl Seen { + /// The page's payments with how each changed since the last page, newest first as the + /// API lists them. A payment without an ID cannot be followed and is left out. + pub fn update(&mut self, page: &Value) -> Vec<(String, Value, Change)> { + let first = !std::mem::replace(&mut self.started, true); + let mut changes = Vec::new(); + for payment in items(page) { + let Some(id) = payment.get("id").and_then(Value::as_str) else { + continue; + }; + let state = state(payment); + let change = match self.states.insert(id.to_owned(), state.clone()) { + _ if first => Change::Snapshot, + None => Change::New, + Some(previous) if previous != state => Change::Updated, + Some(_) => continue, + }; + changes.push((id.to_owned(), payment.clone(), change)); + } + changes + } +} + +/// The page with at most `rows` payments, and how many it left out. +pub fn first(page: &Value, rows: Option) -> (Value, usize) { + let all = items(page); + let Some(rows) = rows.filter(|rows| *rows < all.len()) else { + return (page.clone(), 0); + }; + let kept = Value::Array(all[..rows].to_vec()); + let page = match page { + Value::Object(map) => { + let mut map = map.clone(); + map.insert("items".into(), kept); + Value::Object(map) + } + _ => kept, + }; + (page, all.len() - rows) +} + +/// A bare array or a page's `items`. +pub fn items(page: &Value) -> &[Value] { + page.as_array() + .or_else(|| page.get("items").and_then(Value::as_array)) + .map_or(&[], Vec::as_slice) +} + +/// What marks a payment as changed: its status and when it was last updated. +fn state(payment: &Value) -> String { + let field = |name: &str| payment.get(name).map(Value::to_string).unwrap_or_default(); + format!("{} {}", field("status"), field("updated_at")) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::error::ErrorKind; + use serde_json::json; + + #[test] + fn a_watch_follows_the_newest_updates_whatever_the_query_asked() { + let mut query = vec![ + ("wallet_id".to_owned(), "w".to_owned()), + ("sort_key".to_owned(), "created_at".to_owned()), + ("sort_order".to_owned(), "ASC".to_owned()), + ("cursor".to_owned(), "c1".to_owned()), + ]; + follow_updates(&mut query); + let pairs: Vec<(&str, &str)> = query + .iter() + .map(|(n, v)| (n.as_str(), v.as_str())) + .collect(); + assert_eq!( + pairs, + [ + ("wallet_id", "w"), + ("sort_key", "updated_at"), + ("sort_order", "DESC") + ] + ); + } + + #[test] + fn a_page_is_cut_to_the_rows_that_fit() { + let page = json!({"items": [{"id": "a"}, {"id": "b"}, {"id": "c"}], "next_cursor": "x"}); + let (cut, more) = first(&page, Some(2)); + assert_eq!((items(&cut).len(), more), (2, 1)); + assert_eq!(cut["next_cursor"], "x"); + assert_eq!(first(&page, None).1, 0); + assert_eq!(first(&page, Some(5)).1, 0); + let (bare, more) = first(&json!([1, 2, 3]), Some(1)); + assert_eq!((bare, more), (json!([1]), 2)); + } + + #[test] + fn only_failures_the_server_may_get_past_are_retried() { + let status = |status: u16| Error { + response_status: Some(status), + ..Error::new(ErrorKind::for_http_status(status), "x") + }; + assert!(retries(&Error::transport("reset"))); + for code in [429, 500, 502, 503] { + assert!(retries(&status(code)), "{code}"); + } + for code in [400, 401, 403, 404] { + assert!(!retries(&status(code)), "{code}"); + } + assert!(!retries(&Error::api("no status"))); + } + + #[test] + fn the_first_page_is_a_snapshot_then_only_new_and_changed_payments_count() { + let mut seen = Seen::default(); + let first = json!({"items": [ + {"id": "a", "status": "receiving", "updated_at": "t1"}, + {"status": "completed"} + ]}); + let snapshot = seen.update(&first); + assert_eq!(snapshot.len(), 1); + assert_eq!( + (snapshot[0].0.as_str(), snapshot[0].2), + ("a", Change::Snapshot) + ); + assert!(seen.update(&first).is_empty()); + let next = json!([ + {"id": "b", "status": "receiving"}, + {"id": "a", "status": "completed", "updated_at": "t2"} + ]); + let changes: Vec<_> = seen + .update(&next) + .into_iter() + .map(|(id, _, change)| (id, change.as_str())) + .collect(); + assert_eq!(changes, [("b".into(), "new"), ("a".into(), "updated")]); + assert!(items(&json!({"other": 1})).is_empty()); + } +} diff --git a/tests/cli_integration.rs b/tests/cli_integration.rs index 823a893..b1534b9 100644 --- a/tests/cli_integration.rs +++ b/tests/cli_integration.rs @@ -6217,6 +6217,403 @@ async fn a_qr_code_for_a_ready_invoice_comes_last_so_it_stays_on_screen() { server.verify().await; } +/// Two reads of a payments list: payment A receiving; then B new and A completed. +#[cfg(unix)] +async fn watched_payments(server: &MockServer) { + let a = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa"; + let b = "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb"; + route("GET", payments_path()) + .respond_with(in_order(vec![ + ok(json!({"items": [{"id": a, "status": "receiving", "updated_at": "2026-10-20T00:00:00Z"}]})), + ok(json!({"items": [ + {"id": b, "status": "receiving", "updated_at": "2026-10-20T00:00:01Z"}, + {"id": a, "status": "completed", "updated_at": "2026-10-20T00:00:02Z"} + ]})), + ])) + .mount(server) + .await; +} + +#[cfg(unix)] +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_redirected_watch_writes_one_event_per_new_or_changed_payment() { + use std::io::{BufRead, BufReader}; + use std::process::Stdio; + let (server, dir) = fixture().await; + watched_payments(&server).await; + let mut child = process_at(dir.path(), &server.uri()) + .args([ + "payments", + "list", + "--org", + ORG, + "--env", + ENV, + "--watch", + "--interval", + "1", + ]) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + let stdout = child.stdout.take().unwrap(); + let events = tokio::task::spawn_blocking(move || { + BufReader::new(stdout) + .lines() + .take(3) + .map(|line| serde_json::from_str::(&line.unwrap()).unwrap()) + .collect::>() + }); + let events = tokio::time::timeout(Duration::from_secs(10), events) + .await + .unwrap() + .unwrap(); + let summary: Vec<(String, String, String)> = events + .iter() + .map(|event| { + ( + event["event"].as_str().unwrap().to_owned(), + event["id"].as_str().unwrap().to_owned(), + event["data"]["status"].as_str().unwrap().to_owned(), + ) + }) + .collect(); + assert_eq!( + summary, + [ + ( + "snapshot".into(), + "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa".into(), + "receiving".into() + ), + ( + "new".into(), + "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb".into(), + "receiving".into() + ), + ( + "updated".into(), + "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa".into(), + "completed".into() + ), + ] + ); + assert!(events.iter().all(|event| event["outcome"] == "event")); + interrupt(&child); + let output = tokio::task::spawn_blocking(move || child.wait_with_output().unwrap()) + .await + .unwrap(); + assert_eq!(output.status.code(), Some(130)); + assert_eq!( + String::from_utf8_lossy(&output.stderr), + "Stopped watching payments.\n" + ); +} + +#[cfg(unix)] +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_watch_retries_a_server_error_and_ends_on_a_client_error() { + let (server, dir) = fixture().await; + let a = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa"; + // Every refresh asks for the most recently updated first. A gateway's HTML 502 and a + // 429 with `Retry-After` are retried; the change after them still arrives; a 403 ends it. + route("GET", payments_path()) + .and(query_param("sort_key", "updated_at")) + .and(query_param("sort_order", "DESC")) + .respond_with(in_order(vec![ + ok(json!({"items": [{"id": a, "status": "receiving"}]})), + ResponseTemplate::new(502).set_body_string("Bad Gateway"), + ResponseTemplate::new(429).insert_header("Retry-After", "1"), + ok(json!({"items": [{"id": a, "status": "completed"}]})), + ResponseTemplate::new(403).set_body_json(json!({"error": {"detail": "Forbidden"}})), + ])) + .expect(5) + .mount(&server) + .await; + let result = cli(dir.path(), &server) + .args([ + "payments", + "list", + "--org", + ORG, + "--env", + ENV, + "--watch", + "--interval", + "1", + ]) + .timeout(Duration::from_secs(20)) + .assert() + .code(3); + let statuses: Vec = stdout(&result) + .lines() + .map(|line| serde_json::from_str::(line).unwrap()) + .map(|event| { + format!( + "{} {}", + event["event"].as_str().unwrap(), + event["data"]["status"].as_str().unwrap() + ) + }) + .collect(); + assert_eq!(statuses, ["snapshot receiving", "updated completed"]); + server.verify().await; +} + +/// What a pseudo-terminal has shown so far, and whether it has closed. +#[cfg(unix)] +#[derive(Default)] +struct Shown { + bytes: Vec, + closed: bool, +} + +/// A pseudo-terminal's output, read on its own thread as it arrives. +#[cfg(unix)] +#[derive(Clone, Default)] +struct Drained { + shown: Arc>, + changed: Arc, +} + +#[cfg(unix)] +impl Drained { + /// Block until `ready` holds for the output so far, the terminal closes, or `limit` passes. + fn until(&self, limit: Duration, ready: impl Fn(&str) -> bool) { + let guard = self.shown.lock().unwrap(); + let _ = self + .changed + .wait_timeout_while(guard, limit, |shown| { + !shown.closed && !ready(&String::from_utf8_lossy(&shown.bytes)) + }) + .unwrap(); + } + + /// Everything the terminal showed, once the process has closed it. + fn finish(&self) -> String { + self.until(Duration::from_secs(10), |_| false); + String::from_utf8_lossy(&self.shown.lock().unwrap().bytes).into_owned() + } +} + +#[cfg(unix)] +fn drain(mut controller: std::fs::File) -> Drained { + use std::io::Read; + let drained = Drained::default(); + let writer = drained.clone(); + std::thread::spawn(move || { + let mut chunk = [0; 4096]; + // Linux PTYs return EIO rather than EOF when the terminal end closes. + while let Ok(read @ 1..) = controller.read(&mut chunk) { + writer + .shown + .lock() + .unwrap() + .bytes + .extend_from_slice(&chunk[..read]); + writer.changed.notify_all(); + } + writer.shown.lock().unwrap().closed = true; + writer.changed.notify_all(); + }); + drained +} + +#[cfg(unix)] +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_watch_on_a_saved_login_exchanges_again_when_its_token_expires() { + let (server, dir) = fixture().await; + let _settings = seeded_login(dir.path(), &server, false); + // The first organization token is rejected on the second read; a new exchange follows. + organization_exchange(&server, LOGIN_ACCESS, ORG, "org-token-1", 1).await; + organization_exchange(&server, LOGIN_ACCESS, ORG, "org-token-2", 1).await; + let a = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa"; + route("GET", payments_path()) + .and(header("authorization", "Bearer org-token-1")) + .respond_with(in_order(vec![ + ok(json!({"items": [{"id": a, "status": "receiving"}]})), + ResponseTemplate::new(401).set_body_json(json!({"error": {"detail": "expired"}})), + ])) + .expect(2) + .mount(&server) + .await; + route("GET", payments_path()) + .and(header("authorization", "Bearer org-token-2")) + .respond_with(ok(json!({"items": [{"id": a, "status": "completed"}]}))) + .mount(&server) + .await; + let mut child = process_at(dir.path(), &server.uri()) + .env_remove("VOLTAGE_API_KEY") + .args([ + "payments", + "list", + "--org", + ORG, + "--env", + ENV, + "--account", + LOGIN_NAME, + "--watch", + "--interval", + "1", + ]) + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::piped()) + .spawn() + .unwrap(); + let stdout = child.stdout.take().unwrap(); + let events = tokio::task::spawn_blocking(move || { + use std::io::{BufRead, BufReader}; + BufReader::new(stdout) + .lines() + .take(2) + .map(|line| serde_json::from_str::(&line.unwrap()).unwrap()) + .map(|event| { + format!( + "{} {}", + event["event"].as_str().unwrap(), + event["data"]["status"].as_str().unwrap() + ) + }) + .collect::>() + }); + let events = tokio::time::timeout(Duration::from_secs(15), events) + .await + .unwrap() + .unwrap(); + assert_eq!(events, ["snapshot receiving", "updated completed"]); + interrupt(&child); + let output = tokio::task::spawn_blocking(move || child.wait_with_output().unwrap()) + .await + .unwrap(); + assert_eq!(output.status.code(), Some(130)); + server.verify().await; +} + +#[cfg(unix)] +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_live_watch_that_ends_in_an_error_puts_the_screen_back_first() { + let (server, dir) = fixture().await; + route("GET", payments_path()) + .respond_with(in_order(vec![ + ok(json!({"items": []})), + ResponseTemplate::new(403).set_body_json(json!({"error": {"detail": "Forbidden"}})), + ])) + .mount(&server) + .await; + let args = [ + "payments", + "list", + "--org", + ORG, + "--env", + ENV, + "--watch", + "--interval", + "1", + ]; + let (controller, terminal) = open_pty(); + let child = human_process_at(dir.path(), &server.uri()) + .env("TERM", "xterm-256color") + .args(args) + .stdin(std::process::Stdio::null()) + .stdout(std::process::Stdio::from(terminal.try_clone().unwrap())) + .stderr(std::process::Stdio::from(terminal)) + .spawn() + .unwrap(); + let shown = drain(controller); + let output = tokio::task::spawn_blocking(move || child.wait_with_output().unwrap()) + .await + .unwrap(); + assert_eq!(output.status.code(), Some(3)); + let raw = shown.finish(); + let left = raw.find("\u{1b}[?25h\u{1b}[?1049l").expect(&raw); + let error = raw.find("error:").expect(&raw); + assert!(left < error, "{raw:?}"); +} + +#[cfg(unix)] +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_watch_on_a_terminal_is_a_live_table_that_marks_changes() { + use std::process::Stdio; + let (server, dir) = fixture().await; + watched_payments(&server).await; + let (controller, terminal) = open_pty(); + let child = human_process_at(dir.path(), &server.uri()) + .env("TERM", "xterm-256color") + .env("LANG", "C.UTF-8") + .env_remove("LC_ALL") + .env_remove("LC_CTYPE") + .env("COLUMNS", "200") + .args([ + "payments", + "list", + "--org", + ORG, + "--env", + ENV, + "--watch", + "--interval", + "1", + ]) + .stdin(Stdio::null()) + .stdout(Stdio::from(terminal.try_clone().unwrap())) + .stderr(Stdio::from(terminal)) + .spawn() + .unwrap(); + // Stop once three frames are drawn: the snapshot, B new and A changed, then a quiet one. + let shown = drain(controller); + let wait = shown.clone(); + tokio::task::spawn_blocking(move || { + wait.until(Duration::from_secs(10), |text| { + text.matches("Watching payments").count() >= 3 + }) + }) + .await + .unwrap(); + interrupt(&child); + let output = tokio::task::spawn_blocking(move || child.wait_with_output().unwrap()) + .await + .unwrap(); + assert_eq!(output.status.code(), Some(130)); + let raw = shown.finish(); + // The alternate screen is entered, then left with the cursor shown again. + let entered = raw.find("\u{1b}[?1049h\u{1b}[?25l").expect(&raw); + let left = raw.find("\u{1b}[?25h\u{1b}[?1049l").expect(&raw); + assert!(entered < left, "{raw:?}"); + assert!( + raw[left..].contains("Stopped watching payments."), + "{raw:?}" + ); + let shown = strip_styles(&raw); + assert!( + shown.contains("Watching payments · every 1s · Ctrl-C to stop"), + "{shown}" + ); + // Each refresh is a frame. The snapshot marks nothing; the frame where B first appears + // marks B as new and A as changed; later frames, with nothing new, mark nothing. + let frames: Vec<&str> = shown.split("Watching payments").skip(1).collect(); + assert!(!frames[0].contains('●'), "{}", frames[0]); + let changed = frames + .iter() + .position(|frame| frame.contains("bbbbbbbb")) + .expect(&shown); + assert!(changed > 0, "{shown}"); + let frame = frames[changed]; + assert!( + frame.contains("● bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb"), + "{frame}" + ); + assert!( + frame.contains("● aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa"), + "{frame}" + ); + for later in &frames[changed + 1..] { + assert!(!later.contains('●'), "{later}"); + } +} + #[cfg(unix)] #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn a_failed_payment_on_a_terminal_ends_with_a_cross_before_the_error() {