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

logbrew_cli/
lib.rs
