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