Skip to main content

logbrew_cli/
lib.rs

1//! Native `LogBrew` command-line interface library.
2//!
3//! The CLI is intentionally small and predictable so coding agents can learn it
4//! quickly: `read`, `watch`, `explain`, and `set` cover data access and state
5//! changes, while `login`, `setup`, and `status` cover local configuration.
6
7#![forbid(unsafe_code)]
8
9#[doc(hidden)]
10pub mod auth;
11#[doc(hidden)]
12pub mod auth_namespace;
13mod error;
14#[doc(hidden)]
15pub mod flags;
16pub mod help;
17#[doc(hidden)]
18pub mod ids;
19mod parser;
20#[doc(hidden)]
21pub mod render;
22#[doc(hidden)]
23pub mod setup;
24#[doc(hidden)]
25pub mod status;
26#[doc(hidden)]
27pub mod version;
28
29use auth::{open_browser, resolve_credential, write_logout_result};
30pub use error::{CliError, RuntimeError, write_cli_error, write_runtime_error};
31use futures_util::StreamExt as _;
32pub use parser::parse_command;
33use render::write_api_success;
34use setup::write_setup_plan;
35use status::execute_status;
36use tokio_tungstenite::connect_async;
37use tokio_tungstenite::tungstenite::{Error as WebSocketError, Message};
38use version::execute_version;
39
40/// Initial delay before reconnecting a live watch stream.
41const WATCH_RECONNECT_INITIAL_DELAY: std::time::Duration = std::time::Duration::from_secs(1);
42/// Maximum delay before reconnecting a live watch stream.
43const WATCH_RECONNECT_MAX_DELAY: std::time::Duration = std::time::Duration::from_secs(30);
44/// Maximum jitter added to reconnect delays.
45const WATCH_RECONNECT_JITTER_MAX_MILLIS: u64 = 250;
46
47/// Accepted issue status values for generic recovery text.
48pub(crate) const ISSUE_STATUS_VALUES_NEXT_STEP: &str =
49    "use one of unresolved/open, resolved/closed, ignored";
50/// Accepted issue status values for read filter recovery text.
51pub(crate) const ISSUE_STATUS_FILTER_NEXT_STEP: &str =
52    "use --status unresolved/open, --status resolved/closed, or --status ignored";
53/// Accepted issue status values for missing mutation arguments.
54pub(crate) const ISSUE_STATUS_ARGUMENT_NEXT_STEP: &str =
55    "provide one of unresolved/open, resolved/closed, ignored";
56
57/// Parsed `LogBrew` command.
58#[derive(Debug, Clone, PartialEq, Eq)]
59pub enum Command {
60    /// Shows command usage.
61    Help {
62        /// Help topic to display.
63        topic: HelpTopic,
64        /// Emit machine-readable JSON.
65        json: bool,
66    },
67    /// Opens browser-based authentication.
68    Login {
69        /// Try to open the login URL in the default browser.
70        open_browser: bool,
71        /// Emit machine-readable JSON.
72        json: bool,
73    },
74    /// Removes the local CLI token.
75    Logout {
76        /// Emit machine-readable JSON.
77        json: bool,
78    },
79    /// Detects the current project and prints a non-mutating SDK setup plan.
80    Setup {
81        /// Let the CLI pick the framework or runtime automatically.
82        auto: bool,
83        /// Suppress confirmation prompts.
84        yes: bool,
85        /// Emit machine-readable JSON.
86        json: bool,
87    },
88    /// Checks local auth and server reachability.
89    Status {
90        /// Emit machine-readable JSON.
91        json: bool,
92    },
93    /// Prints the installed CLI version.
94    Version {
95        /// Emit machine-readable JSON.
96        json: bool,
97    },
98    /// Reads historical observability data.
99    Read {
100        /// Resource to read.
101        target: ReadTarget,
102        /// Read filters.
103        options: Box<ReadOptions>,
104        /// Emit machine-readable JSON.
105        json: bool,
106    },
107    /// Watches live observability data.
108    Watch {
109        /// Resource to watch.
110        target: WatchTarget,
111        /// Live watch filters applied client-side.
112        options: WatchOptions,
113        /// Emit machine-readable JSON.
114        json: bool,
115    },
116    /// Fetches context for an issue or trace so an agent can explain it.
117    Explain {
118        /// Resource to explain.
119        target: ExplainTarget,
120        /// Emit machine-readable JSON.
121        json: bool,
122    },
123    /// Mutates server-side state.
124    Set {
125        /// Target state mutation.
126        target: SetTarget,
127        /// Emit machine-readable JSON.
128        json: bool,
129    },
130}
131
132/// Help topic for CLI usage output.
133#[derive(Debug, Clone, Copy, PartialEq, Eq)]
134pub enum HelpTopic {
135    /// Root command overview.
136    Root,
137    /// Browser login command.
138    Login,
139    /// Local logout command.
140    Logout,
141    /// SDK setup command.
142    Setup,
143    /// Status check command.
144    Status,
145    /// Installed CLI version command.
146    Version,
147    /// Authentication workflow overview.
148    Auth,
149    /// Machine-readable output overview.
150    Json,
151    /// First-run examples and common workflows.
152    Examples,
153    /// Read command overview.
154    Read,
155    /// Log reading command.
156    ReadLogs,
157    /// Issue reading command.
158    ReadIssues,
159    /// Action reading command.
160    ReadActions,
161    /// Release reading command.
162    ReadReleases,
163    /// Trace reading command.
164    ReadTrace,
165    /// Single issue reading command.
166    ReadIssue,
167    /// Live watch command.
168    Watch,
169    /// Explain command.
170    Explain,
171    /// State mutation command.
172    Set,
173}
174
175impl HelpTopic {
176    /// Returns a stable machine-readable topic name.
177    #[must_use]
178    pub const fn key(self) -> &'static str {
179        match self {
180            Self::Root => "root",
181            Self::Login => "login",
182            Self::Logout => "logout",
183            Self::Setup => "setup",
184            Self::Status => "status",
185            Self::Version => "version",
186            Self::Auth => "auth",
187            Self::Json => "json",
188            Self::Examples => "examples",
189            Self::Read => "read",
190            Self::ReadLogs => "read_logs",
191            Self::ReadIssues => "read_issues",
192            Self::ReadActions => "read_actions",
193            Self::ReadReleases => "read_releases",
194            Self::ReadTrace => "read_trace",
195            Self::ReadIssue => "read_issue",
196            Self::Watch => "watch",
197            Self::Explain => "explain",
198            Self::Set => "set",
199        }
200    }
201}
202
203/// Historical data target for `read`.
204#[derive(Debug, Clone, PartialEq, Eq)]
205pub enum ReadTarget {
206    /// Structured logs.
207    Logs,
208    /// Grouped issues.
209    Issues,
210    /// Product actions.
211    Actions,
212    /// Release summaries.
213    Releases,
214    /// One trace by ID.
215    Trace(String),
216    /// One issue by ID.
217    Issue(String),
218}
219
220/// Filters for historical read commands.
221#[derive(Debug, Clone, Default, PartialEq, Eq)]
222pub struct ReadOptions {
223    /// Optional action name filter.
224    pub name: Option<String>,
225    /// Optional relative or absolute lower time bound.
226    pub since: Option<String>,
227    /// Optional user or actor filter.
228    pub user: Option<String>,
229    /// Optional trace ID filter.
230    pub trace: Option<String>,
231    /// Optional log severity filter.
232    pub level: Option<String>,
233    /// Optional log message substring search.
234    pub search: Option<String>,
235    /// Optional project filter.
236    pub project: Option<String>,
237    /// Optional release filter.
238    pub release: Option<String>,
239    /// Optional environment filter.
240    pub environment: Option<String>,
241    /// Optional issue status filter.
242    pub status: Option<String>,
243    /// Optional row limit.
244    pub limit: Option<String>,
245}
246
247impl ReadOptions {
248    /// Returns the first filter that trace-detail reads cannot apply.
249    #[must_use]
250    pub(crate) fn first_trace_detail_unsupported_flag(&self) -> Option<&'static str> {
251        first_present_flag([
252            (self.name.is_some(), "--name"),
253            (self.since.is_some(), "--since"),
254            (self.user.is_some(), "--user"),
255            (self.trace.is_some(), "--trace"),
256            (self.level.is_some(), "--severity"),
257            (self.search.is_some(), "--search"),
258            (self.status.is_some(), "--status"),
259            (self.limit.is_some(), "--limit"),
260        ])
261    }
262
263    /// Returns the first filter that issue-detail reads cannot apply.
264    #[must_use]
265    pub(crate) fn first_issue_detail_unsupported_flag(&self) -> Option<&'static str> {
266        first_present_flag([
267            (self.name.is_some(), "--name"),
268            (self.since.is_some(), "--since"),
269            (self.user.is_some(), "--user"),
270            (self.trace.is_some(), "--trace"),
271            (self.level.is_some(), "--severity"),
272            (self.search.is_some(), "--search"),
273            (self.project.is_some(), "--project"),
274            (self.release.is_some(), "--release"),
275            (self.environment.is_some(), "--environment"),
276            (self.status.is_some(), "--status"),
277            (self.limit.is_some(), "--limit"),
278        ])
279    }
280
281    /// Returns the first filter that log reads cannot apply.
282    #[must_use]
283    pub(crate) fn first_log_unsupported_flag(&self) -> Option<&'static str> {
284        first_present_flag([
285            (self.name.is_some(), "--name"),
286            (self.user.is_some(), "--user"),
287            (self.status.is_some(), "--status"),
288        ])
289    }
290
291    /// Returns the first filter that issue list reads cannot apply.
292    #[must_use]
293    pub(crate) fn first_issue_list_unsupported_flag(&self) -> Option<&'static str> {
294        first_present_flag([
295            (self.name.is_some(), "--name"),
296            (self.since.is_some(), "--since"),
297            (self.user.is_some(), "--user"),
298            (self.trace.is_some(), "--trace"),
299            (self.level.is_some(), "--severity"),
300            (self.search.is_some(), "--search"),
301        ])
302    }
303
304    /// Returns the first filter that action reads cannot apply.
305    #[must_use]
306    pub(crate) fn first_action_unsupported_flag(&self) -> Option<&'static str> {
307        first_present_flag([
308            (self.trace.is_some(), "--trace"),
309            (self.level.is_some(), "--severity"),
310            (self.search.is_some(), "--search"),
311            (self.status.is_some(), "--status"),
312        ])
313    }
314
315    /// Returns the first filter that release reads cannot apply.
316    #[must_use]
317    pub(crate) fn first_release_unsupported_flag(&self) -> Option<&'static str> {
318        first_present_flag([
319            (self.name.is_some(), "--name"),
320            (self.since.is_some(), "--since"),
321            (self.user.is_some(), "--user"),
322            (self.trace.is_some(), "--trace"),
323            (self.level.is_some(), "--severity"),
324            (self.search.is_some(), "--search"),
325            (self.status.is_some(), "--status"),
326        ])
327    }
328}
329
330/// Returns the first present flag in declaration order.
331fn first_present_flag<const N: usize>(flags: [(bool, &'static str); N]) -> Option<&'static str> {
332    flags
333        .iter()
334        .find_map(|(present, flag)| present.then_some(*flag))
335}
336
337/// Live stream target for `watch`.
338#[derive(Debug, Clone, Copy, PartialEq, Eq)]
339pub enum WatchTarget {
340    /// All supported live event types.
341    All,
342    /// Structured logs.
343    Logs,
344    /// Grouped issues.
345    Issues,
346    /// Product actions.
347    Actions,
348}
349
350/// Client-side filters for live watch commands.
351#[derive(Debug, Clone, Default, PartialEq, Eq)]
352pub struct WatchOptions {
353    /// Canonical severity filters for logs and issues.
354    pub severity: Vec<String>,
355}
356
357/// Context target for `explain`.
358#[derive(Debug, Clone, PartialEq, Eq)]
359pub enum ExplainTarget {
360    /// One issue by ID.
361    Issue(String),
362    /// One trace by ID.
363    Trace(String),
364}
365
366/// Mutation target for `set`.
367#[derive(Debug, Clone, PartialEq, Eq)]
368pub enum SetTarget {
369    /// Update one issue status.
370    IssueStatus {
371        /// Issue identifier.
372        id: String,
373        /// New issue status.
374        status: String,
375    },
376}
377
378/// Process environment needed by the CLI.
379#[derive(Debug, Clone, PartialEq, Eq)]
380pub struct CliEnvironment {
381    /// Base API URL.
382    pub base_url: String,
383    /// Optional bearer token.
384    pub token: Option<String>,
385    /// Optional home directory.
386    pub home: Option<std::path::PathBuf>,
387    /// Optional current working directory.
388    pub cwd: Option<std::path::PathBuf>,
389}
390
391impl CliEnvironment {
392    /// Loads CLI environment from process variables.
393    #[must_use]
394    pub fn from_process() -> Self {
395        Self {
396            base_url: std::env::var("LOGBREW_API_URL")
397                .unwrap_or_else(|_| String::from("https://api.logbrew.co")),
398            token: std::env::var("LOGBREW_TOKEN").ok(),
399            home: std::env::var_os("HOME").map(std::path::PathBuf::from),
400            cwd: std::env::current_dir().ok(),
401        }
402    }
403}
404
405impl Command {
406    /// Returns the HTTP API path for commands backed by a single REST request.
407    #[must_use]
408    pub fn http_path(&self) -> Option<String> {
409        match self {
410            Self::Read {
411                target, options, ..
412            } => Some(read_path(
413                target,
414                &ReadPathFilters {
415                    name: options.name.as_deref(),
416                    since: options.since.as_deref(),
417                    user: options.user.as_deref(),
418                    trace: options.trace.as_deref(),
419                    level: options.level.as_deref(),
420                    search: options.search.as_deref(),
421                    project: options.project.as_deref(),
422                    release: options.release.as_deref(),
423                    environment: options.environment.as_deref(),
424                    status: options.status.as_deref(),
425                    limit: options.limit.as_deref(),
426                },
427            )),
428            Self::Explain { target, .. } => Some(explain_path(target)),
429            Self::Set { target, .. } => Some(set_path(target)),
430            Self::Help { .. }
431            | Self::Login { .. }
432            | Self::Logout { .. }
433            | Self::Setup { .. }
434            | Self::Status { .. }
435            | Self::Version { .. }
436            | Self::Watch { .. } => None,
437        }
438    }
439
440    /// Returns whether command output should be JSON.
441    #[must_use]
442    pub const fn wants_json(&self) -> bool {
443        match self {
444            Self::Help { json, .. }
445            | Self::Login { json, .. }
446            | Self::Logout { json }
447            | Self::Status { json }
448            | Self::Version { json }
449            | Self::Read { json, .. }
450            | Self::Watch { json, .. }
451            | Self::Explain { json, .. }
452            | Self::Set { json, .. }
453            | Self::Setup { json, .. } => *json,
454        }
455    }
456
457    /// Returns the HTTP method for commands backed by a REST request.
458    #[must_use]
459    pub const fn http_method(&self) -> Option<HttpMethod> {
460        match self {
461            Self::Read { .. } | Self::Explain { .. } => Some(HttpMethod::Get),
462            Self::Set { .. } => Some(HttpMethod::Patch),
463            Self::Help { .. }
464            | Self::Login { .. }
465            | Self::Logout { .. }
466            | Self::Setup { .. }
467            | Self::Status { .. }
468            | Self::Version { .. }
469            | Self::Watch { .. } => None,
470        }
471    }
472
473    /// Returns JSON request body for mutation commands.
474    #[must_use]
475    pub fn request_body(&self) -> Option<serde_json::Value> {
476        match self {
477            Self::Set {
478                target: SetTarget::IssueStatus { status, .. },
479                ..
480            } => Some(serde_json::json!({ "status": status })),
481            Self::Help { .. }
482            | Self::Login { .. }
483            | Self::Logout { .. }
484            | Self::Setup { .. }
485            | Self::Status { .. }
486            | Self::Version { .. }
487            | Self::Read { .. }
488            | Self::Watch { .. }
489            | Self::Explain { .. } => None,
490        }
491    }
492}
493
494/// HTTP method used by a CLI command.
495#[derive(Debug, Clone, Copy, PartialEq, Eq)]
496pub enum HttpMethod {
497    /// GET request.
498    Get,
499    /// PATCH request.
500    Patch,
501}
502
503/// Executes a parsed command.
504///
505/// # Errors
506///
507/// Returns [`RuntimeError`] if output, browser launch, auth, or HTTP fails.
508pub async fn execute_command<W: std::io::Write>(
509    command: &Command,
510    env: &CliEnvironment,
511    output: &mut W,
512) -> Result<(), RuntimeError> {
513    match command {
514        Command::Help { topic, json } => execute_help(*topic, *json, output),
515        Command::Login { open_browser, json } => execute_login(env, *open_browser, *json, output),
516        Command::Logout { json } => execute_logout(env, *json, output),
517        Command::Setup { auto, yes, json } => execute_setup(env, *auto, *yes, *json, output),
518        Command::Status { json } => execute_status(env, *json, output).await,
519        Command::Version { json } => execute_version(*json, output),
520        Command::Read { .. } | Command::Explain { .. } | Command::Set { .. } => {
521            execute_http(command, env, output).await
522        }
523        Command::Watch {
524            target,
525            options,
526            json,
527        } => execute_watch(env, *target, options, *json, output).await,
528    }
529}
530
531/// Emits CLI help.
532fn execute_help<W: std::io::Write>(
533    topic: HelpTopic,
534    json: bool,
535    output: &mut W,
536) -> Result<(), RuntimeError> {
537    let help = help::help_text(topic);
538    if json {
539        let body = serde_json::json!({
540            "ok": true,
541            "topic": topic.key(),
542            "help": help,
543        });
544        writeln!(output, "{body}")?;
545    } else {
546        writeln!(output, "{help}")?;
547    }
548    Ok(())
549}
550
551/// Executes browser login bootstrap.
552fn execute_login<W: std::io::Write>(
553    env: &CliEnvironment,
554    should_open_browser: bool,
555    json: bool,
556    output: &mut W,
557) -> Result<(), RuntimeError> {
558    let auth_url = format!("{}/api/auth/cli/login", env.base_url.trim_end_matches('/'));
559    let opened = should_open_browser && open_browser(auth_url.as_str());
560
561    if json {
562        let body = serde_json::json!({
563            "ok": true,
564            "auth_url": auth_url,
565            "browser_opened": opened,
566            "next": "open auth_url in a browser",
567        });
568        writeln!(output, "{body}")?;
569    } else {
570        writeln!(output, "Open this URL to log in: {auth_url}")?;
571        writeln!(
572            output,
573            "Browser: {}",
574            if opened { "opened" } else { "not opened" }
575        )?;
576        writeln!(output, "Next: open the URL in a browser")?;
577    }
578    Ok(())
579}
580
581/// Executes local logout.
582fn execute_logout<W: std::io::Write>(
583    env: &CliEnvironment,
584    json: bool,
585    output: &mut W,
586) -> Result<(), RuntimeError> {
587    write_logout_result(env, json, output)?;
588    Ok(())
589}
590
591/// Executes setup planning.
592fn execute_setup<W: std::io::Write>(
593    env: &CliEnvironment,
594    auto: bool,
595    yes: bool,
596    json: bool,
597    output: &mut W,
598) -> Result<(), RuntimeError> {
599    write_setup_plan(env.cwd.as_deref(), auto, yes, json, output)?;
600    Ok(())
601}
602
603/// Executes commands backed by one HTTP request.
604async fn execute_http<W: std::io::Write>(
605    command: &Command,
606    env: &CliEnvironment,
607    output: &mut W,
608) -> Result<(), RuntimeError> {
609    let path = command.http_path().ok_or(CliError::UnknownCommand)?;
610    let url = format!("{}{}", env.base_url.trim_end_matches('/'), path);
611    let client = reqwest::Client::builder()
612        .timeout(std::time::Duration::from_secs(30))
613        .connect_timeout(std::time::Duration::from_secs(10))
614        .build()?;
615
616    let mut request = match command.http_method().unwrap_or(HttpMethod::Get) {
617        HttpMethod::Get => client.get(url),
618        HttpMethod::Patch => client.patch(url),
619    };
620
621    let credential = resolve_credential(env)?;
622    request = request.bearer_auth(credential.token);
623
624    if let Some(body) = command.request_body() {
625        request = request.json(&body);
626    }
627
628    let response = request.send().await?;
629    let status = response.status();
630    let body = response.text().await?;
631
632    if !status.is_success() {
633        return Err(RuntimeError::Api {
634            status: status.as_u16(),
635            body,
636            auth_source: credential.source,
637            auth_label: credential.label,
638        });
639    }
640
641    write_api_success(command, body.as_str(), output)?;
642    Ok(())
643}
644
645/// Executes the public live WebSocket watch flow.
646async fn execute_watch<W: std::io::Write>(
647    env: &CliEnvironment,
648    target: WatchTarget,
649    options: &WatchOptions,
650    json: bool,
651    output: &mut W,
652) -> Result<(), RuntimeError> {
653    if !json {
654        return Err(RuntimeError::Unavailable {
655            message: "watch streams JSON for agents",
656            next: "run logbrew watch --json",
657        });
658    }
659
660    let credential = resolve_credential(env)?;
661    let mut reconnect_backoff = WatchReconnectBackoff::default();
662    loop {
663        let ticket = match request_feed_ticket(env, &credential).await {
664            Ok(ticket) => ticket,
665            Err(error) if reconnect_backoff.connected_once() && !runtime_error_is_auth(&error) => {
666                tokio::time::sleep(reconnect_backoff.next_delay()).await;
667                continue;
668            }
669            Err(error) => return Err(error),
670        };
671        let live_url = feed_live_url(env.base_url.as_str(), ticket.as_str())?;
672        let (mut websocket, _) = match connect_async(live_url.as_str()).await {
673            Ok(connection) => connection,
674            Err(error)
675                if reconnect_backoff.connected_once() && !websocket_error_is_auth(&error) =>
676            {
677                tokio::time::sleep(reconnect_backoff.next_delay()).await;
678                continue;
679            }
680            Err(error) => return Err(map_websocket_connect_error(error)),
681        };
682        reconnect_backoff.mark_connected();
683
684        let mut emitted_before_disconnect = false;
685        loop {
686            let Some(message) = websocket.next().await else {
687                break;
688            };
689            let message = match message {
690                Ok(message) => message,
691                Err(error) if websocket_error_is_auth(&error) => {
692                    return Err(map_websocket_stream_error(error));
693                }
694                Err(_) => break,
695            };
696            match message {
697                Message::Text(text) => {
698                    let event = parse_live_event(text.as_str())?;
699                    if watch_event_matches(target, options, &event) {
700                        writeln!(output, "{event}")?;
701                    }
702                    emitted_before_disconnect = true;
703                }
704                Message::Binary(_) | Message::Ping(_) | Message::Pong(_) | Message::Frame(_) => {}
705                Message::Close(_) => return Ok(()),
706            }
707        }
708        if emitted_before_disconnect {
709            reconnect_backoff.reset();
710        }
711        tokio::time::sleep(reconnect_backoff.next_delay()).await;
712    }
713}
714
715/// Reconnect state for long-running live watch streams.
716#[derive(Debug, Default)]
717struct WatchReconnectBackoff {
718    /// Whether a live WebSocket connection has ever been established.
719    connected_once: bool,
720    /// Consecutive reconnect attempts since the last stable event.
721    attempts: u32,
722}
723
724impl WatchReconnectBackoff {
725    /// Returns whether the stream has connected at least once.
726    const fn connected_once(&self) -> bool {
727        self.connected_once
728    }
729
730    /// Records a successful WebSocket connection.
731    const fn mark_connected(&mut self) {
732        self.connected_once = true;
733    }
734
735    /// Resets retry delay after a stream successfully emits data.
736    const fn reset(&mut self) {
737        self.attempts = 0;
738    }
739
740    /// Returns the next capped exponential reconnect delay.
741    fn next_delay(&mut self) -> std::time::Duration {
742        let exponent = self.attempts.min(5);
743        let multiplier = 1_u64 << exponent;
744        self.attempts = self.attempts.saturating_add(1);
745        let base = WATCH_RECONNECT_INITIAL_DELAY
746            .as_secs()
747            .saturating_mul(multiplier)
748            .min(WATCH_RECONNECT_MAX_DELAY.as_secs());
749        let delay = std::time::Duration::from_secs(base) + watch_reconnect_jitter();
750        if delay > WATCH_RECONNECT_MAX_DELAY {
751            WATCH_RECONNECT_MAX_DELAY
752        } else {
753            delay
754        }
755    }
756}
757
758/// Returns small jitter for reconnect delays without adding a random dependency.
759fn watch_reconnect_jitter() -> std::time::Duration {
760    let Ok(elapsed) = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH) else {
761        return std::time::Duration::ZERO;
762    };
763    std::time::Duration::from_millis(
764        u64::from(elapsed.subsec_millis()) % WATCH_RECONNECT_JITTER_MAX_MILLIS,
765    )
766}
767
768/// Returns whether a runtime error should stop watch reconnect attempts.
769const fn runtime_error_is_auth(error: &RuntimeError) -> bool {
770    matches!(
771        error,
772        RuntimeError::MissingToken | RuntimeError::Api { status: 401, .. }
773    )
774}
775
776/// Returns whether a WebSocket error is an auth failure.
777fn websocket_error_is_auth(error: &WebSocketError) -> bool {
778    matches!(error, WebSocketError::Http(response) if response.status().as_u16() == 401)
779}
780
781/// Requests a short-lived WebSocket feed ticket from the public API.
782async fn request_feed_ticket(
783    env: &CliEnvironment,
784    credential: &auth::AuthCredential,
785) -> Result<String, RuntimeError> {
786    let url = format!("{}/api/feed/ticket", env.base_url.trim_end_matches('/'));
787    let client = reqwest::Client::builder()
788        .timeout(std::time::Duration::from_secs(30))
789        .connect_timeout(std::time::Duration::from_secs(10))
790        .build()?;
791    let response = client
792        .post(url)
793        .bearer_auth(credential.token.as_str())
794        .send()
795        .await?;
796    let status = response.status();
797    let body = response.text().await?;
798    if !status.is_success() {
799        return Err(RuntimeError::Api {
800            status: status.as_u16(),
801            body,
802            auth_source: credential.source,
803            auth_label: credential.label,
804        });
805    }
806
807    let value = serde_json::from_str::<serde_json::Value>(body.as_str()).map_err(|_| {
808        RuntimeError::Unavailable {
809            message: "feed ticket response was not valid JSON",
810            next: "retry logbrew watch or run logbrew status",
811        }
812    })?;
813    value
814        .get("ticket")
815        .and_then(serde_json::Value::as_str)
816        .map(str::trim)
817        .filter(|ticket| !ticket.is_empty())
818        .map(ToOwned::to_owned)
819        .ok_or(RuntimeError::Unavailable {
820            message: "feed ticket response did not include a ticket",
821            next: "retry logbrew watch or run logbrew status",
822        })
823}
824
825/// Builds the WebSocket live feed URL without exposing the opaque ticket elsewhere.
826fn feed_live_url(base_url: &str, ticket: &str) -> Result<String, RuntimeError> {
827    let trimmed = base_url.trim_end_matches('/');
828    let (scheme, rest) = websocket_base_parts(trimmed).ok_or(RuntimeError::Unavailable {
829        message: "LOGBREW_API_URL must start with http:// or https://",
830        next: "check LOGBREW_API_URL or run logbrew status",
831    })?;
832    Ok(format!(
833        "{scheme}://{rest}/api/feed/live?ticket={}",
834        encode_component(ticket)
835    ))
836}
837
838/// Converts an HTTP API base URL into WebSocket scheme and authority/path base parts.
839fn websocket_base_parts(base_url: &str) -> Option<(&'static str, &str)> {
840    base_url
841        .strip_prefix("https://")
842        .map(|rest| ("wss", rest))
843        .or_else(|| base_url.strip_prefix("http://").map(|rest| ("ws", rest)))
844}
845
846/// Parses one backend live event object.
847fn parse_live_event(text: &str) -> Result<serde_json::Value, RuntimeError> {
848    serde_json::from_str::<serde_json::Value>(text).map_err(|_| RuntimeError::Unavailable {
849        message: "live watch event was not valid JSON",
850        next: "retry logbrew watch or check LOGBREW_API_URL",
851    })
852}
853
854/// Returns whether an event should be emitted for the requested watch target and filters.
855fn watch_event_matches(
856    target: WatchTarget,
857    options: &WatchOptions,
858    event: &serde_json::Value,
859) -> bool {
860    target_matches_event(target, event) && severity_matches(options, event)
861}
862
863/// Returns whether the event type belongs to the selected target.
864fn target_matches_event(target: WatchTarget, event: &serde_json::Value) -> bool {
865    let event_type = event
866        .get("type")
867        .and_then(serde_json::Value::as_str)
868        .unwrap_or_default();
869    match target {
870        WatchTarget::All => true,
871        WatchTarget::Logs => event_type == "native_log",
872        WatchTarget::Issues => event_type == "native_issue",
873        WatchTarget::Actions => event_type == "native_action",
874    }
875}
876
877/// Applies client-side severity filters to log and issue events.
878fn severity_matches(options: &WatchOptions, event: &serde_json::Value) -> bool {
879    if options.severity.is_empty() {
880        return true;
881    }
882    let Some(severity) = event
883        .get("data")
884        .and_then(|data| data.get("severity").or_else(|| data.get("level")))
885        .and_then(serde_json::Value::as_str)
886    else {
887        return false;
888    };
889    options
890        .severity
891        .iter()
892        .any(|allowed| allowed.as_str() == severity)
893}
894
895/// Maps a WebSocket connection failure to a token-safe runtime error.
896fn map_websocket_connect_error(error: WebSocketError) -> RuntimeError {
897    match error {
898        WebSocketError::Http(response) if response.status().as_u16() == 401 => {
899            RuntimeError::Unavailable {
900                message: "live watch ticket was rejected",
901                next: "run logbrew login",
902            }
903        }
904        WebSocketError::Http(_) => RuntimeError::Unavailable {
905            message: "live watch websocket upgrade failed",
906            next: "retry logbrew watch or check LOGBREW_API_URL",
907        },
908        WebSocketError::ConnectionClosed
909        | WebSocketError::AlreadyClosed
910        | WebSocketError::Io(_)
911        | WebSocketError::Tls(_)
912        | WebSocketError::Capacity(_)
913        | WebSocketError::Protocol(_)
914        | WebSocketError::WriteBufferFull(_)
915        | WebSocketError::Utf8(_)
916        | WebSocketError::AttackAttempt
917        | WebSocketError::Url(_)
918        | WebSocketError::HttpFormat(_) => RuntimeError::Unavailable {
919            message: "live watch websocket failed",
920            next: "retry logbrew watch or check LOGBREW_API_URL",
921        },
922    }
923}
924
925/// Maps an established WebSocket stream failure to a token-safe runtime error.
926fn map_websocket_stream_error(error: WebSocketError) -> RuntimeError {
927    match error {
928        WebSocketError::ConnectionClosed | WebSocketError::AlreadyClosed => {
929            RuntimeError::Unavailable {
930                message: "live watch websocket closed",
931                next: "retry logbrew watch",
932            }
933        }
934        WebSocketError::Http(response) if response.status().as_u16() == 401 => {
935            RuntimeError::Unavailable {
936                message: "live watch ticket was rejected",
937                next: "run logbrew login",
938            }
939        }
940        WebSocketError::Http(_)
941        | WebSocketError::Io(_)
942        | WebSocketError::Tls(_)
943        | WebSocketError::Capacity(_)
944        | WebSocketError::Protocol(_)
945        | WebSocketError::WriteBufferFull(_)
946        | WebSocketError::Utf8(_)
947        | WebSocketError::AttackAttempt
948        | WebSocketError::Url(_)
949        | WebSocketError::HttpFormat(_) => RuntimeError::Unavailable {
950            message: "live watch websocket failed",
951            next: "retry logbrew watch or check LOGBREW_API_URL",
952        },
953    }
954}
955
956/// Read endpoint filter values.
957struct ReadPathFilters<'a> {
958    /// Optional action name filter.
959    name: Option<&'a str>,
960    /// Optional lower time bound.
961    since: Option<&'a str>,
962    /// Optional user or actor filter.
963    user: Option<&'a str>,
964    /// Optional trace ID filter.
965    trace: Option<&'a str>,
966    /// Optional log severity filter.
967    level: Option<&'a str>,
968    /// Optional log message substring search.
969    search: Option<&'a str>,
970    /// Optional project filter.
971    project: Option<&'a str>,
972    /// Optional release filter.
973    release: Option<&'a str>,
974    /// Optional environment filter.
975    environment: Option<&'a str>,
976    /// Optional issue status filter.
977    status: Option<&'a str>,
978    /// Optional row limit.
979    limit: Option<&'a str>,
980}
981
982/// Builds a read endpoint path.
983fn read_path(target: &ReadTarget, filters: &ReadPathFilters<'_>) -> String {
984    match target {
985        ReadTarget::Logs => path_with_query(
986            "/api/logs",
987            &[
988                ("severity", filters.level),
989                ("search", filters.search),
990                ("since", filters.since),
991                ("trace_id", filters.trace),
992                ("project_id", filters.project),
993                ("release", filters.release),
994                ("environment", filters.environment),
995                ("limit", filters.limit),
996            ],
997        ),
998        ReadTarget::Issues => path_with_query(
999            "/api/telemetry/issues",
1000            &[
1001                ("status", filters.status),
1002                ("project_id", filters.project),
1003                ("release", filters.release),
1004                ("environment", filters.environment),
1005                ("limit", filters.limit),
1006            ],
1007        ),
1008        ReadTarget::Actions => path_with_query(
1009            "/api/telemetry/actions",
1010            &[
1011                ("name", filters.name),
1012                ("since", filters.since),
1013                ("distinct_id", filters.user),
1014                ("project_id", filters.project),
1015                ("release", filters.release),
1016                ("environment", filters.environment),
1017                ("limit", filters.limit),
1018            ],
1019        ),
1020        ReadTarget::Releases => path_with_query(
1021            "/api/telemetry/releases",
1022            &[
1023                ("project_id", filters.project),
1024                ("release", filters.release),
1025                ("environment", filters.environment),
1026                ("limit", filters.limit),
1027            ],
1028        ),
1029        ReadTarget::Trace(id) => path_with_query(
1030            &format!("/api/telemetry/traces/{}", encode_component(id)),
1031            &[
1032                ("project_id", filters.project),
1033                ("release", filters.release),
1034                ("environment", filters.environment),
1035            ],
1036        ),
1037        ReadTarget::Issue(id) => format!("/api/telemetry/issues/{}", encode_component(id)),
1038    }
1039}
1040
1041/// Builds an explain endpoint path.
1042fn explain_path(target: &ExplainTarget) -> String {
1043    match target {
1044        ExplainTarget::Issue(id) => format!("/api/telemetry/issues/{}", encode_component(id)),
1045        ExplainTarget::Trace(id) => format!("/api/telemetry/traces/{}", encode_component(id)),
1046    }
1047}
1048
1049/// Builds a mutation endpoint path.
1050fn set_path(target: &SetTarget) -> String {
1051    match target {
1052        SetTarget::IssueStatus { id, .. } => {
1053            format!("/api/telemetry/issues/{}", encode_component(id))
1054        }
1055    }
1056}
1057
1058/// Builds a path with query parameters.
1059fn path_with_query(path: &str, params: &[(&str, Option<&str>)]) -> String {
1060    let query = params
1061        .iter()
1062        .filter_map(|(name, value)| value.map(|v| format!("{name}={}", encode_component(v))))
1063        .collect::<Vec<_>>();
1064
1065    if query.is_empty() {
1066        path.to_owned()
1067    } else {
1068        format!("{path}?{}", query.join("&"))
1069    }
1070}
1071
1072/// Percent-encodes a path or query component without adding a dependency.
1073fn encode_component(value: &str) -> String {
1074    let mut encoded = String::new();
1075    for byte in value.bytes() {
1076        if byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.' | b'~') {
1077            encoded.push(char::from(byte));
1078        } else {
1079            encoded.push('%');
1080            encoded.push(hex_digit(byte >> 4));
1081            encoded.push(hex_digit(byte & 0x0f));
1082        }
1083    }
1084    encoded
1085}
1086
1087/// Converts a nibble to an uppercase hexadecimal digit.
1088fn hex_digit(nibble: u8) -> char {
1089    match nibble {
1090        0..=9 => char::from(b'0' + nibble),
1091        10..=15 => char::from(b'A' + (nibble - 10)),
1092        _ => '?',
1093    }
1094}