1#![forbid(unsafe_code)]
8
9#[doc(hidden)]
10pub mod auth;
11#[doc(hidden)]
12pub mod auth_namespace;
13#[doc(hidden)]
14pub mod doctor;
15mod error;
16#[doc(hidden)]
17pub mod flags;
18pub mod help;
19#[doc(hidden)]
20pub mod ids;
21#[doc(hidden)]
22pub mod investigate;
23mod native_debug_artifacts;
24mod parser;
25mod project_create;
26#[doc(hidden)]
27pub mod render;
28#[doc(hidden)]
29pub mod setup;
30#[doc(hidden)]
31pub mod status;
32mod support;
33mod usage;
34#[doc(hidden)]
35pub mod version;
36
37use auth::{AuthCredential, execute_login, execute_logout, send_authenticated_with_refresh};
38pub use error::{CliError, RuntimeError, write_cli_error, write_runtime_error};
39use futures_util::StreamExt as _;
40pub use parser::parse_command;
41use render::write_api_success;
42use setup::write_setup_plan;
43use status::execute_status;
44use tokio_tungstenite::connect_async;
45use tokio_tungstenite::tungstenite::{Error as WebSocketError, Message};
46use version::execute_version;
47
48const WATCH_RECONNECT_INITIAL_DELAY: std::time::Duration = std::time::Duration::from_secs(1);
50const WATCH_RECONNECT_MAX_DELAY: std::time::Duration = std::time::Duration::from_secs(30);
52const WATCH_RECONNECT_JITTER_MAX_MILLIS: u64 = 250;
54
55pub(crate) const ISSUE_STATUS_VALUES_NEXT_STEP: &str =
57 "use one of unresolved/open, resolved/closed, ignored";
58pub(crate) const ISSUE_STATUS_FILTER_NEXT_STEP: &str =
60 "use --status unresolved/open, --status resolved/closed, or --status ignored";
61pub(crate) const ISSUE_STATUS_ARGUMENT_NEXT_STEP: &str =
63 "provide one of unresolved/open, resolved/closed, ignored";
64
65#[derive(Debug, Clone, PartialEq, Eq)]
67pub enum Command {
68 Help {
70 topic: HelpTopic,
72 json: bool,
74 },
75 Login {
77 open_browser: bool,
79 json: bool,
81 },
82 Logout {
84 json: bool,
86 },
87 Setup {
89 auto: bool,
91 yes: bool,
93 json: bool,
95 },
96 Status {
98 json: bool,
100 },
101 Doctor {
103 project_id: String,
105 json: bool,
107 },
108 ProjectCreate {
110 options: ProjectCreateOptions,
112 json: bool,
114 },
115 Usage {
117 json: bool,
119 },
120 Version {
122 json: bool,
124 },
125 Read {
127 target: ReadTarget,
129 options: Box<ReadOptions>,
131 json: bool,
133 },
134 Watch {
136 target: WatchTarget,
138 options: WatchOptions,
140 json: bool,
142 },
143 Explain {
145 target: ExplainTarget,
147 json: bool,
149 },
150 InvestigateIssue {
152 issue_id: String,
154 json: bool,
156 },
157 NativeDebugArtifacts {
159 target: NativeDebugArtifactsTarget,
161 json: bool,
163 },
164 Set {
166 target: SetTarget,
168 json: bool,
170 },
171 ProjectSetupSeen {
173 project_id: String,
175 options: ProjectSetupSeenOptions,
177 json: bool,
179 },
180 Support {
182 target: SupportTarget,
184 json: bool,
186 },
187}
188
189#[derive(Debug, Clone, Copy, PartialEq, Eq)]
191pub enum HelpTopic {
192 Root,
194 Login,
196 Logout,
198 Setup,
200 Status,
202 Version,
204 Auth,
206 Json,
208 Examples,
210 Projects,
212 Usage,
214 Read,
216 ReadLogs,
218 ReadIssues,
220 ReadActions,
222 ReadReleases,
224 ReadTraces,
226 ReadTrace,
228 ReadIssue,
230 Watch,
232 Explain,
234 Investigate,
236 NativeDebugArtifacts,
238 Set,
240 Support,
242}
243
244impl HelpTopic {
245 #[must_use]
247 pub const fn key(self) -> &'static str {
248 match self {
249 Self::Root => "root",
250 Self::Login => "login",
251 Self::Logout => "logout",
252 Self::Setup => "setup",
253 Self::Status => "status",
254 Self::Version => "version",
255 Self::Auth => "auth",
256 Self::Json => "json",
257 Self::Examples => "examples",
258 Self::Projects => "projects",
259 Self::Usage => "usage",
260 Self::Read => "read",
261 Self::ReadLogs => "read_logs",
262 Self::ReadIssues => "read_issues",
263 Self::ReadActions => "read_actions",
264 Self::ReadReleases => "read_releases",
265 Self::ReadTraces => "read_traces",
266 Self::ReadTrace => "read_trace",
267 Self::ReadIssue => "read_issue",
268 Self::Watch => "watch",
269 Self::Explain => "explain",
270 Self::Investigate => "investigate",
271 Self::NativeDebugArtifacts => "debug_artifacts",
272 Self::Set => "set",
273 Self::Support => "support",
274 }
275 }
276}
277
278#[derive(Debug, Clone, PartialEq, Eq)]
280pub enum ReadTarget {
281 Logs,
283 Issues,
285 Actions,
287 Releases,
289 Traces,
291 Trace(String),
293 Issue(String),
295}
296
297#[derive(Debug, Clone, Default, PartialEq, Eq)]
299pub struct ReadOptions {
300 pub name: Option<String>,
302 pub service: Option<String>,
304 pub since: Option<String>,
306 pub user: Option<String>,
308 pub trace: Option<String>,
310 pub level: Option<String>,
312 pub search: Option<String>,
314 pub project: Option<String>,
316 pub release: Option<String>,
318 pub environment: Option<String>,
320 pub status: Option<String>,
322 pub limit: Option<String>,
324 pub min_duration_ms: Option<String>,
326 pub pagination: Option<String>,
328 pub cursor_time: Option<String>,
330 pub cursor_id: Option<String>,
332}
333
334impl ReadOptions {
335 #[must_use]
337 pub(crate) fn first_trace_detail_unsupported_flag(&self) -> Option<&'static str> {
338 first_present_flag([
339 (self.name.is_some(), "--name"),
340 (self.service.is_some(), "--service"),
341 (self.since.is_some(), "--since"),
342 (self.user.is_some(), "--user"),
343 (self.trace.is_some(), "--trace"),
344 (self.level.is_some(), "--severity"),
345 (self.search.is_some(), "--search"),
346 (self.status.is_some(), "--status"),
347 (self.limit.is_some(), "--limit"),
348 (self.min_duration_ms.is_some(), "--min-duration-ms"),
349 (self.pagination.is_some(), "--pagination"),
350 (self.cursor_time.is_some(), "--cursor-time"),
351 (self.cursor_id.is_some(), "--cursor-id"),
352 ])
353 }
354
355 #[must_use]
357 pub(crate) fn first_issue_detail_unsupported_flag(&self) -> Option<&'static str> {
358 first_present_flag([
359 (self.name.is_some(), "--name"),
360 (self.service.is_some(), "--service"),
361 (self.since.is_some(), "--since"),
362 (self.user.is_some(), "--user"),
363 (self.trace.is_some(), "--trace"),
364 (self.level.is_some(), "--severity"),
365 (self.search.is_some(), "--search"),
366 (self.project.is_some(), "--project"),
367 (self.release.is_some(), "--release"),
368 (self.environment.is_some(), "--environment"),
369 (self.status.is_some(), "--status"),
370 (self.limit.is_some(), "--limit"),
371 (self.min_duration_ms.is_some(), "--min-duration-ms"),
372 (self.pagination.is_some(), "--pagination"),
373 (self.cursor_time.is_some(), "--cursor-time"),
374 (self.cursor_id.is_some(), "--cursor-id"),
375 ])
376 }
377
378 #[must_use]
380 pub(crate) fn first_log_unsupported_flag(&self) -> Option<&'static str> {
381 first_present_flag([
382 (self.name.is_some(), "--name"),
383 (self.user.is_some(), "--user"),
384 (self.status.is_some(), "--status"),
385 (self.min_duration_ms.is_some(), "--min-duration-ms"),
386 ])
387 }
388
389 #[must_use]
391 pub(crate) fn first_issue_list_unsupported_flag(&self) -> Option<&'static str> {
392 first_present_flag([
393 (self.name.is_some(), "--name"),
394 (self.user.is_some(), "--user"),
395 (self.trace.is_some(), "--trace"),
396 (self.level.is_some(), "--severity"),
397 (self.search.is_some(), "--search"),
398 (self.min_duration_ms.is_some(), "--min-duration-ms"),
399 ])
400 }
401
402 #[must_use]
404 pub(crate) fn first_action_unsupported_flag(&self) -> Option<&'static str> {
405 first_present_flag([
406 (self.trace.is_some(), "--trace"),
407 (self.level.is_some(), "--severity"),
408 (self.search.is_some(), "--search"),
409 (self.status.is_some(), "--status"),
410 (self.min_duration_ms.is_some(), "--min-duration-ms"),
411 ])
412 }
413
414 #[must_use]
416 pub(crate) fn first_release_unsupported_flag(&self) -> Option<&'static str> {
417 first_present_flag([
418 (self.name.is_some(), "--name"),
419 (self.user.is_some(), "--user"),
420 (self.trace.is_some(), "--trace"),
421 (self.level.is_some(), "--severity"),
422 (self.search.is_some(), "--search"),
423 (self.status.is_some(), "--status"),
424 (self.min_duration_ms.is_some(), "--min-duration-ms"),
425 (self.pagination.is_some(), "--pagination"),
426 (self.cursor_time.is_some(), "--cursor-time"),
427 (self.cursor_id.is_some(), "--cursor-id"),
428 ])
429 }
430
431 #[must_use]
433 pub(crate) fn first_trace_list_unsupported_flag(&self) -> Option<&'static str> {
434 first_present_flag([
435 (self.name.is_some(), "--name"),
436 (self.user.is_some(), "--user"),
437 (self.trace.is_some(), "--trace"),
438 (self.level.is_some(), "--severity"),
439 (self.search.is_some(), "--search"),
440 (self.pagination.is_some(), "--pagination"),
441 (self.cursor_time.is_some(), "--cursor-time"),
442 (self.cursor_id.is_some(), "--cursor-id"),
443 ])
444 }
445}
446
447fn first_present_flag<const N: usize>(flags: [(bool, &'static str); N]) -> Option<&'static str> {
449 flags
450 .iter()
451 .find_map(|(present, flag)| present.then_some(*flag))
452}
453
454#[derive(Debug, Clone, Copy, PartialEq, Eq)]
456pub enum WatchTarget {
457 All,
459 Logs,
461 Issues,
463 Actions,
465}
466
467#[derive(Debug, Clone, Default, PartialEq, Eq)]
469pub struct WatchOptions {
470 pub severity: Vec<String>,
472}
473
474#[derive(Debug, Clone, Default, PartialEq, Eq)]
476pub struct ProjectSetupSeenOptions {
477 pub runtime: Option<String>,
479 pub source: Option<String>,
481 pub environment: Option<String>,
483}
484
485#[derive(Debug, Clone, PartialEq, Eq)]
487pub struct ProjectCreateOptions {
488 pub name: String,
490 pub runtime: Option<String>,
492 pub environment: Option<String>,
494 pub ingest_key_file: String,
496 pub abandon_retry: bool,
498}
499
500#[derive(Debug, Clone, PartialEq, Eq)]
502pub enum NativeDebugArtifactsTarget {
503 Upload(NativeDebugUploadOptions),
505 Lookup(NativeDebugLookupOptions),
507}
508
509#[derive(Debug, Clone, PartialEq, Eq)]
511pub struct NativeDebugLookupOptions {
512 pub project_id: String,
514 pub release: String,
516 pub environment: String,
518 pub service: String,
520 pub image_uuid: String,
522 pub architecture: String,
524}
525
526#[derive(Debug, Clone, PartialEq, Eq)]
528pub struct NativeDebugUploadOptions {
529 pub path: String,
531 pub project_id: String,
533 pub release: String,
535 pub environment: String,
537 pub service: String,
539}
540
541#[derive(Debug, Clone, PartialEq, Eq)]
543pub enum SupportTarget {
544 Create(Box<SupportTicketCreateOptions>),
546 List(Box<SupportTicketListOptions>),
548 Detail(String),
550 ContextHistory {
552 ticket_id: String,
554 },
555 ReplyContext(Box<SupportContextReplyOptions>),
557 UpdateStatus {
559 ticket_id: String,
561 status: SupportTicketLifecycleStatus,
563 },
564}
565
566#[derive(Debug, Clone, Default, PartialEq, Eq)]
568pub struct SupportContextReplyOptions {
569 pub ticket_id: String,
571 pub context: String,
573 pub retry_key: String,
575 pub diagnostics: bool,
577}
578
579#[derive(Debug, Clone, Copy, PartialEq, Eq)]
581pub enum SupportTicketLifecycleStatus {
582 Open,
584 Closed,
586}
587
588impl SupportTicketLifecycleStatus {
589 #[must_use]
591 pub const fn as_str(self) -> &'static str {
592 match self {
593 Self::Open => "open",
594 Self::Closed => "closed",
595 }
596 }
597}
598
599#[derive(Debug, Clone, Default, PartialEq, Eq)]
601pub struct SupportTicketCreateOptions {
602 pub category: String,
604 pub title: String,
606 pub description: String,
608 pub project_id: Option<String>,
610 pub environment: Option<String>,
612 pub runtime: Option<String>,
614 pub framework: Option<String>,
616 pub sdk_package: Option<String>,
618 pub sdk_version: Option<String>,
620 pub release: Option<String>,
622 pub trace_id: Option<String>,
624 pub event_id: Option<String>,
626 pub diagnostics: bool,
628}
629
630#[derive(Debug, Clone, Default, PartialEq, Eq)]
632pub struct SupportTicketListOptions {
633 pub project_id: Option<String>,
635 pub status: Option<String>,
637 pub source: Option<String>,
639 pub category: Option<String>,
641 pub release: Option<String>,
643 pub limit: Option<String>,
645 pub pagination: Option<String>,
647 pub cursor_time: Option<String>,
649 pub cursor_id: Option<String>,
651}
652
653#[derive(Debug, Clone, PartialEq, Eq)]
655pub enum ExplainTarget {
656 Issue(String),
658 Trace(String),
660}
661
662#[derive(Debug, Clone, PartialEq, Eq)]
664pub enum SetTarget {
665 IssueStatus {
667 id: String,
669 status: String,
671 },
672}
673
674#[derive(Debug, Clone, PartialEq, Eq)]
676pub struct CliEnvironment {
677 pub base_url: String,
679 pub token: Option<String>,
681 pub home: Option<std::path::PathBuf>,
683 pub cwd: Option<std::path::PathBuf>,
685}
686
687impl CliEnvironment {
688 #[must_use]
690 pub fn from_process() -> Self {
691 Self {
692 base_url: std::env::var("LOGBREW_API_URL")
693 .unwrap_or_else(|_| String::from("https://api.logbrew.co")),
694 token: std::env::var("LOGBREW_TOKEN").ok(),
695 home: std::env::var_os("HOME").map(std::path::PathBuf::from),
696 cwd: std::env::current_dir().ok(),
697 }
698 }
699}
700
701impl Command {
702 #[must_use]
704 pub fn http_path(&self) -> Option<String> {
705 match self {
706 Self::Read {
707 target, options, ..
708 } => Some(read_path(
709 target,
710 &ReadPathFilters {
711 name: options.name.as_deref(),
712 service: options.service.as_deref(),
713 since: options.since.as_deref(),
714 user: options.user.as_deref(),
715 trace: options.trace.as_deref(),
716 level: options.level.as_deref(),
717 search: options.search.as_deref(),
718 project: options.project.as_deref(),
719 release: options.release.as_deref(),
720 environment: options.environment.as_deref(),
721 status: options.status.as_deref(),
722 limit: options.limit.as_deref(),
723 min_duration_ms: options.min_duration_ms.as_deref(),
724 pagination: options.pagination.as_deref(),
725 cursor_time: options.cursor_time.as_deref(),
726 cursor_id: options.cursor_id.as_deref(),
727 },
728 )),
729 Self::Explain { target, .. } => Some(explain_path(target)),
730 Self::Set { target, .. } => Some(set_path(target)),
731 Self::ProjectSetupSeen { project_id, .. } => {
732 Some(format!("/api/projects/{project_id}/setup/seen"))
733 }
734 Self::ProjectCreate { .. } => Some(String::from("/api/projects")),
735 Self::Support { target, .. } => Some(support::path(target)),
736 Self::Help { .. }
737 | Self::Login { .. }
738 | Self::Logout { .. }
739 | Self::Setup { .. }
740 | Self::Status { .. }
741 | Self::Doctor { .. }
742 | Self::Usage { .. }
743 | Self::Version { .. }
744 | Self::InvestigateIssue { .. }
745 | Self::NativeDebugArtifacts { .. }
746 | Self::Watch { .. } => None,
747 }
748 }
749
750 #[must_use]
752 pub const fn wants_json(&self) -> bool {
753 match self {
754 Self::Help { json, .. }
755 | Self::Login { json, .. }
756 | Self::Logout { json }
757 | Self::Status { json }
758 | Self::Doctor { json, .. }
759 | Self::ProjectCreate { json, .. }
760 | Self::Usage { json }
761 | Self::Version { json }
762 | Self::Read { json, .. }
763 | Self::Watch { json, .. }
764 | Self::Explain { json, .. }
765 | Self::InvestigateIssue { json, .. }
766 | Self::NativeDebugArtifacts { json, .. }
767 | Self::Set { json, .. }
768 | Self::ProjectSetupSeen { json, .. }
769 | Self::Support { json, .. }
770 | Self::Setup { json, .. } => *json,
771 }
772 }
773
774 #[must_use]
776 pub const fn http_method(&self) -> Option<HttpMethod> {
777 match self {
778 Self::ProjectCreate { .. }
779 | Self::ProjectSetupSeen { .. }
780 | Self::Support {
781 target: SupportTarget::Create(_),
782 ..
783 }
784 | Self::Support {
785 target: SupportTarget::ReplyContext(_),
786 ..
787 } => Some(HttpMethod::Post),
788 Self::Support {
789 target: SupportTarget::UpdateStatus { .. },
790 ..
791 }
792 | Self::Set { .. } => Some(HttpMethod::Patch),
793 Self::Read { .. } | Self::Explain { .. } | Self::Support { .. } => {
794 Some(HttpMethod::Get)
795 }
796 Self::Help { .. }
797 | Self::Login { .. }
798 | Self::Logout { .. }
799 | Self::Setup { .. }
800 | Self::Status { .. }
801 | Self::Doctor { .. }
802 | Self::Usage { .. }
803 | Self::Version { .. }
804 | Self::InvestigateIssue { .. }
805 | Self::NativeDebugArtifacts { .. }
806 | Self::Watch { .. } => None,
807 }
808 }
809
810 #[must_use]
812 pub fn request_body(&self) -> Option<serde_json::Value> {
813 self.request_body_for_token(None)
814 }
815
816 #[must_use]
818 fn request_body_for_token(&self, token: Option<&str>) -> Option<serde_json::Value> {
819 match self {
820 Self::Set {
821 target: SetTarget::IssueStatus { status, .. },
822 ..
823 } => Some(serde_json::json!({ "status": status })),
824 Self::ProjectSetupSeen { options, .. } => Some(project_setup_seen_body(options, token)),
825 Self::ProjectCreate { options, .. } => Some(project_create_body(options)),
826 Self::Support {
827 target: SupportTarget::Create(options),
828 ..
829 } => Some(support::create_body(options)),
830 Self::Support {
831 target: SupportTarget::ReplyContext(options),
832 ..
833 } => Some(support::context_body(options)),
834 Self::Support {
835 target: SupportTarget::UpdateStatus { status, .. },
836 ..
837 } => Some(serde_json::json!({"status": status.as_str()})),
838 Self::Help { .. }
839 | Self::Login { .. }
840 | Self::Logout { .. }
841 | Self::Setup { .. }
842 | Self::Status { .. }
843 | Self::Doctor { .. }
844 | Self::Usage { .. }
845 | Self::Version { .. }
846 | Self::Read { .. }
847 | Self::Watch { .. }
848 | Self::Explain { .. }
849 | Self::InvestigateIssue { .. }
850 | Self::NativeDebugArtifacts { .. }
851 | Self::Support { .. } => None,
852 }
853 }
854
855 fn idempotency_key(&self) -> Option<&str> {
857 match self {
858 Self::Support {
859 target: SupportTarget::ReplyContext(options),
860 ..
861 } => Some(options.retry_key.as_str()),
862 Self::Help { .. }
863 | Self::Login { .. }
864 | Self::Logout { .. }
865 | Self::Setup { .. }
866 | Self::Status { .. }
867 | Self::Doctor { .. }
868 | Self::Usage { .. }
869 | Self::Version { .. }
870 | Self::Read { .. }
871 | Self::Watch { .. }
872 | Self::Explain { .. }
873 | Self::InvestigateIssue { .. }
874 | Self::NativeDebugArtifacts { .. }
875 | Self::Set { .. }
876 | Self::ProjectSetupSeen { .. }
877 | Self::ProjectCreate { .. }
878 | Self::Support { .. } => None,
879 }
880 }
881}
882
883#[derive(Debug, Clone, Copy, PartialEq, Eq)]
885pub enum HttpMethod {
886 Get,
888 Post,
890 Patch,
892}
893
894fn project_setup_seen_body(
896 options: &ProjectSetupSeenOptions,
897 token: Option<&str>,
898) -> serde_json::Value {
899 let mut body = serde_json::Map::new();
900 if let Some(runtime) = options.runtime.as_ref() {
901 drop(body.insert(
902 "runtime".to_owned(),
903 serde_json::Value::String(runtime.clone()),
904 ));
905 }
906 if let Some(source) = setup_seen_source(options, token) {
907 drop(body.insert("source".to_owned(), serde_json::Value::String(source)));
908 }
909 if let Some(environment) = options.environment.as_ref() {
910 drop(body.insert(
911 "environment".to_owned(),
912 serde_json::Value::String(environment.clone()),
913 ));
914 }
915 serde_json::Value::Object(body)
916}
917
918fn project_create_body(options: &ProjectCreateOptions) -> serde_json::Value {
920 let mut body = serde_json::Map::new();
921 drop(body.insert(
922 "name".to_owned(),
923 serde_json::Value::String(options.name.clone()),
924 ));
925 if let Some(runtime) = options.runtime.as_ref() {
926 drop(body.insert(
927 "runtime".to_owned(),
928 serde_json::Value::String(runtime.clone()),
929 ));
930 }
931 if let Some(environment) = options.environment.as_ref() {
932 drop(body.insert(
933 "environment".to_owned(),
934 serde_json::Value::String(environment.clone()),
935 ));
936 }
937 drop(body.insert(
938 "source".to_owned(),
939 serde_json::Value::String(String::from("cli")),
940 ));
941 serde_json::Value::Object(body)
942}
943
944fn setup_seen_source(options: &ProjectSetupSeenOptions, token: Option<&str>) -> Option<String> {
946 if token_is_project_ingest_key(token) {
947 return None;
948 }
949 Some(options.source.as_deref().unwrap_or("cli").to_owned())
950}
951
952fn token_is_project_ingest_key(token: Option<&str>) -> bool {
954 token.is_some_and(|token| token.trim_start().starts_with("lbw_ingest_"))
955}
956
957pub async fn execute_command<W: std::io::Write>(
963 command: &Command,
964 env: &CliEnvironment,
965 output: &mut W,
966) -> Result<(), RuntimeError> {
967 match command {
968 Command::Help { topic, json } => execute_help(*topic, *json, output),
969 Command::Login { open_browser, json } => {
970 execute_login(env, *open_browser, *json, output).await
971 }
972 Command::Logout { json } => execute_logout(env, *json, output).await,
973 Command::Setup { auto, yes, json } => execute_setup(env, *auto, *yes, *json, output),
974 Command::Status { json } => execute_status(env, *json, output).await,
975 Command::Doctor { project_id, json } => {
976 doctor::execute(env, project_id.as_str(), *json, output).await
977 }
978 Command::ProjectCreate { options, json } => {
979 project_create::execute(env, options, *json, output).await
980 }
981 Command::Usage { json } => usage::execute(env, *json, output).await,
982 Command::Version { json } => execute_version(*json, output),
983 Command::InvestigateIssue { issue_id, json } => {
984 investigate::execute(env, issue_id.as_str(), *json, output).await
985 }
986 Command::NativeDebugArtifacts { target, json } => {
987 native_debug_artifacts::execute(env, target, *json, output).await
988 }
989 Command::Read { .. }
990 | Command::Explain { .. }
991 | Command::Set { .. }
992 | Command::ProjectSetupSeen { .. }
993 | Command::Support { .. } => execute_http(command, env, output).await,
994 Command::Watch {
995 target,
996 options,
997 json,
998 } => execute_watch(env, *target, options, *json, output).await,
999 }
1000}
1001
1002fn execute_help<W: std::io::Write>(
1004 topic: HelpTopic,
1005 json: bool,
1006 output: &mut W,
1007) -> Result<(), RuntimeError> {
1008 let help = help::help_text(topic);
1009 if json {
1010 let body = serde_json::json!({
1011 "ok": true,
1012 "topic": topic.key(),
1013 "help": help,
1014 });
1015 writeln!(output, "{body}")?;
1016 } else {
1017 writeln!(output, "{help}")?;
1018 }
1019 Ok(())
1020}
1021
1022fn execute_setup<W: std::io::Write>(
1024 env: &CliEnvironment,
1025 auto: bool,
1026 yes: bool,
1027 json: bool,
1028 output: &mut W,
1029) -> Result<(), RuntimeError> {
1030 write_setup_plan(env.cwd.as_deref(), auto, yes, json, output)?;
1031 Ok(())
1032}
1033
1034async fn execute_http<W: std::io::Write>(
1036 command: &Command,
1037 env: &CliEnvironment,
1038 output: &mut W,
1039) -> Result<(), RuntimeError> {
1040 let path = command.http_path().ok_or(CliError::UnknownCommand)?;
1041 let url = format!("{}{}", env.base_url.trim_end_matches('/'), path);
1042 let client = reqwest::Client::builder()
1043 .timeout(std::time::Duration::from_secs(30))
1044 .connect_timeout(std::time::Duration::from_secs(10))
1045 .build()?;
1046
1047 let support_command = matches!(command, Command::Support { .. });
1048 let response_result = send_authenticated_with_refresh(&client, env, |client, credential| {
1049 build_command_request(client, command, url.as_str(), credential)
1050 })
1051 .await;
1052 let (response, credential) = match response_result {
1053 Ok(response) => response,
1054 Err(RuntimeError::Http(_)) if support_command => return Err(support_transport_error()),
1055 Err(error) => return Err(error),
1056 };
1057 let status = response.status();
1058 let body = match response.text().await {
1059 Ok(body) => body,
1060 Err(_) if support_command => return Err(support_transport_error()),
1061 Err(error) => return Err(RuntimeError::Http(error)),
1062 };
1063
1064 if !status.is_success() {
1065 let body = if let Command::Support { target, .. } = command {
1066 support::safe_error_body(target, status.as_u16())
1067 } else {
1068 credential.redact_response_body(body.as_str())
1069 };
1070 return Err(RuntimeError::Api {
1071 status: status.as_u16(),
1072 body,
1073 auth_source: credential.source(),
1074 auth_label: credential.label(),
1075 });
1076 }
1077
1078 write_api_success(command, body.as_str(), output)?;
1079 Ok(())
1080}
1081
1082const fn support_transport_error() -> RuntimeError {
1084 RuntimeError::Unavailable {
1085 message: "support request could not be completed",
1086 next: "check network connectivity and retry the support command",
1087 }
1088}
1089
1090fn build_command_request(
1092 client: &reqwest::Client,
1093 command: &Command,
1094 url: &str,
1095 credential: &AuthCredential,
1096) -> reqwest::RequestBuilder {
1097 let mut request = match command.http_method().unwrap_or(HttpMethod::Get) {
1098 HttpMethod::Get => client.get(url),
1099 HttpMethod::Post => client.post(url),
1100 HttpMethod::Patch => client.patch(url),
1101 }
1102 .bearer_auth(credential.token());
1103 if let Some(body) = command.request_body_for_token(Some(credential.token())) {
1104 request = request.json(&body);
1105 }
1106 if let Some(key) = command.idempotency_key() {
1107 request = request.header("Idempotency-Key", key);
1108 }
1109 request
1110}
1111
1112async fn execute_watch<W: std::io::Write>(
1114 env: &CliEnvironment,
1115 target: WatchTarget,
1116 options: &WatchOptions,
1117 json: bool,
1118 output: &mut W,
1119) -> Result<(), RuntimeError> {
1120 if !json {
1121 return Err(RuntimeError::Unavailable {
1122 message: "watch streams JSON for agents",
1123 next: "run logbrew watch --json",
1124 });
1125 }
1126
1127 let mut reconnect_backoff = WatchReconnectBackoff::default();
1128 loop {
1129 let ticket = match request_feed_ticket(env).await {
1130 Ok(ticket) => ticket,
1131 Err(error) if reconnect_backoff.connected_once() && !runtime_error_is_auth(&error) => {
1132 tokio::time::sleep(reconnect_backoff.next_delay()).await;
1133 continue;
1134 }
1135 Err(error) => return Err(error),
1136 };
1137 let live_url = feed_live_url(env.base_url.as_str(), ticket.as_str())?;
1138 let (mut websocket, _) = match connect_async(live_url.as_str()).await {
1139 Ok(connection) => connection,
1140 Err(error)
1141 if reconnect_backoff.connected_once() && !websocket_error_is_auth(&error) =>
1142 {
1143 tokio::time::sleep(reconnect_backoff.next_delay()).await;
1144 continue;
1145 }
1146 Err(error) => return Err(map_websocket_connect_error(error)),
1147 };
1148 reconnect_backoff.mark_connected();
1149
1150 let mut emitted_before_disconnect = false;
1151 loop {
1152 let Some(message) = websocket.next().await else {
1153 break;
1154 };
1155 let message = match message {
1156 Ok(message) => message,
1157 Err(error) if websocket_error_is_auth(&error) => {
1158 return Err(map_websocket_stream_error(error));
1159 }
1160 Err(_) => break,
1161 };
1162 match message {
1163 Message::Text(text) => {
1164 let event = parse_live_event(text.as_str())?;
1165 if watch_event_matches(target, options, &event) {
1166 writeln!(output, "{event}")?;
1167 }
1168 emitted_before_disconnect = true;
1169 }
1170 Message::Binary(_) | Message::Ping(_) | Message::Pong(_) | Message::Frame(_) => {}
1171 Message::Close(_) => return Ok(()),
1172 }
1173 }
1174 if emitted_before_disconnect {
1175 reconnect_backoff.reset();
1176 }
1177 tokio::time::sleep(reconnect_backoff.next_delay()).await;
1178 }
1179}
1180
1181#[derive(Debug, Default)]
1183struct WatchReconnectBackoff {
1184 connected_once: bool,
1186 attempts: u32,
1188}
1189
1190impl WatchReconnectBackoff {
1191 const fn connected_once(&self) -> bool {
1193 self.connected_once
1194 }
1195
1196 const fn mark_connected(&mut self) {
1198 self.connected_once = true;
1199 }
1200
1201 const fn reset(&mut self) {
1203 self.attempts = 0;
1204 }
1205
1206 fn next_delay(&mut self) -> std::time::Duration {
1208 let exponent = self.attempts.min(5);
1209 let multiplier = 1_u64 << exponent;
1210 self.attempts = self.attempts.saturating_add(1);
1211 let base = WATCH_RECONNECT_INITIAL_DELAY
1212 .as_secs()
1213 .saturating_mul(multiplier)
1214 .min(WATCH_RECONNECT_MAX_DELAY.as_secs());
1215 let delay = std::time::Duration::from_secs(base) + watch_reconnect_jitter();
1216 if delay > WATCH_RECONNECT_MAX_DELAY {
1217 WATCH_RECONNECT_MAX_DELAY
1218 } else {
1219 delay
1220 }
1221 }
1222}
1223
1224fn watch_reconnect_jitter() -> std::time::Duration {
1226 let Ok(elapsed) = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH) else {
1227 return std::time::Duration::ZERO;
1228 };
1229 std::time::Duration::from_millis(
1230 u64::from(elapsed.subsec_millis()) % WATCH_RECONNECT_JITTER_MAX_MILLIS,
1231 )
1232}
1233
1234const fn runtime_error_is_auth(error: &RuntimeError) -> bool {
1236 matches!(
1237 error,
1238 RuntimeError::MissingToken | RuntimeError::Api { status: 401, .. }
1239 )
1240}
1241
1242fn websocket_error_is_auth(error: &WebSocketError) -> bool {
1244 matches!(error, WebSocketError::Http(response) if response.status().as_u16() == 401)
1245}
1246
1247async fn request_feed_ticket(env: &CliEnvironment) -> Result<String, RuntimeError> {
1249 let url = format!("{}/api/feed/ticket", env.base_url.trim_end_matches('/'));
1250 let client = reqwest::Client::builder()
1251 .timeout(std::time::Duration::from_secs(30))
1252 .connect_timeout(std::time::Duration::from_secs(10))
1253 .build()?;
1254 let (response, credential) =
1255 send_authenticated_with_refresh(&client, env, |client, credential| {
1256 client.post(url.as_str()).bearer_auth(credential.token())
1257 })
1258 .await?;
1259 let status = response.status();
1260 let body = response.text().await?;
1261 if !status.is_success() {
1262 return Err(RuntimeError::Api {
1263 status: status.as_u16(),
1264 body: credential.redact_response_body(body.as_str()),
1265 auth_source: credential.source(),
1266 auth_label: credential.label(),
1267 });
1268 }
1269
1270 let value = serde_json::from_str::<serde_json::Value>(body.as_str()).map_err(|_| {
1271 RuntimeError::Unavailable {
1272 message: "feed ticket response was not valid JSON",
1273 next: "retry logbrew watch or run logbrew status",
1274 }
1275 })?;
1276 value
1277 .get("ticket")
1278 .and_then(serde_json::Value::as_str)
1279 .map(str::trim)
1280 .filter(|ticket| !ticket.is_empty())
1281 .map(ToOwned::to_owned)
1282 .ok_or(RuntimeError::Unavailable {
1283 message: "feed ticket response did not include a ticket",
1284 next: "retry logbrew watch or run logbrew status",
1285 })
1286}
1287
1288fn feed_live_url(base_url: &str, ticket: &str) -> Result<String, RuntimeError> {
1290 let trimmed = base_url.trim_end_matches('/');
1291 let (scheme, rest) = websocket_base_parts(trimmed).ok_or(RuntimeError::Unavailable {
1292 message: "LOGBREW_API_URL must start with http:// or https://",
1293 next: "check LOGBREW_API_URL or run logbrew status",
1294 })?;
1295 Ok(format!(
1296 "{scheme}://{rest}/api/feed/live?ticket={}",
1297 encode_component(ticket)
1298 ))
1299}
1300
1301fn websocket_base_parts(base_url: &str) -> Option<(&'static str, &str)> {
1303 base_url
1304 .strip_prefix("https://")
1305 .map(|rest| ("wss", rest))
1306 .or_else(|| base_url.strip_prefix("http://").map(|rest| ("ws", rest)))
1307}
1308
1309fn parse_live_event(text: &str) -> Result<serde_json::Value, RuntimeError> {
1311 serde_json::from_str::<serde_json::Value>(text).map_err(|_| RuntimeError::Unavailable {
1312 message: "live watch event was not valid JSON",
1313 next: "retry logbrew watch or check LOGBREW_API_URL",
1314 })
1315}
1316
1317fn watch_event_matches(
1319 target: WatchTarget,
1320 options: &WatchOptions,
1321 event: &serde_json::Value,
1322) -> bool {
1323 target_matches_event(target, event) && severity_matches(options, event)
1324}
1325
1326fn target_matches_event(target: WatchTarget, event: &serde_json::Value) -> bool {
1328 let event_type = event
1329 .get("type")
1330 .and_then(serde_json::Value::as_str)
1331 .unwrap_or_default();
1332 match target {
1333 WatchTarget::All => true,
1334 WatchTarget::Logs => event_type == "native_log",
1335 WatchTarget::Issues => event_type == "native_issue",
1336 WatchTarget::Actions => event_type == "native_action",
1337 }
1338}
1339
1340fn severity_matches(options: &WatchOptions, event: &serde_json::Value) -> bool {
1342 if options.severity.is_empty() {
1343 return true;
1344 }
1345 let Some(severity) = event
1346 .get("data")
1347 .and_then(|data| data.get("severity").or_else(|| data.get("level")))
1348 .and_then(serde_json::Value::as_str)
1349 else {
1350 return false;
1351 };
1352 options
1353 .severity
1354 .iter()
1355 .any(|allowed| allowed.as_str() == severity)
1356}
1357
1358fn map_websocket_connect_error(error: WebSocketError) -> RuntimeError {
1360 match error {
1361 WebSocketError::Http(response) if response.status().as_u16() == 401 => {
1362 RuntimeError::Unavailable {
1363 message: "live watch ticket was rejected",
1364 next: "run logbrew login",
1365 }
1366 }
1367 WebSocketError::Http(_) => RuntimeError::Unavailable {
1368 message: "live watch websocket upgrade failed",
1369 next: "retry logbrew watch or check LOGBREW_API_URL",
1370 },
1371 WebSocketError::ConnectionClosed
1372 | WebSocketError::AlreadyClosed
1373 | WebSocketError::Io(_)
1374 | WebSocketError::Tls(_)
1375 | WebSocketError::Capacity(_)
1376 | WebSocketError::Protocol(_)
1377 | WebSocketError::WriteBufferFull(_)
1378 | WebSocketError::Utf8(_)
1379 | WebSocketError::AttackAttempt
1380 | WebSocketError::Url(_)
1381 | WebSocketError::HttpFormat(_) => RuntimeError::Unavailable {
1382 message: "live watch websocket failed",
1383 next: "retry logbrew watch or check LOGBREW_API_URL",
1384 },
1385 }
1386}
1387
1388fn map_websocket_stream_error(error: WebSocketError) -> RuntimeError {
1390 match error {
1391 WebSocketError::ConnectionClosed | WebSocketError::AlreadyClosed => {
1392 RuntimeError::Unavailable {
1393 message: "live watch websocket closed",
1394 next: "retry logbrew watch",
1395 }
1396 }
1397 WebSocketError::Http(response) if response.status().as_u16() == 401 => {
1398 RuntimeError::Unavailable {
1399 message: "live watch ticket was rejected",
1400 next: "run logbrew login",
1401 }
1402 }
1403 WebSocketError::Http(_)
1404 | WebSocketError::Io(_)
1405 | WebSocketError::Tls(_)
1406 | WebSocketError::Capacity(_)
1407 | WebSocketError::Protocol(_)
1408 | WebSocketError::WriteBufferFull(_)
1409 | WebSocketError::Utf8(_)
1410 | WebSocketError::AttackAttempt
1411 | WebSocketError::Url(_)
1412 | WebSocketError::HttpFormat(_) => RuntimeError::Unavailable {
1413 message: "live watch websocket failed",
1414 next: "retry logbrew watch or check LOGBREW_API_URL",
1415 },
1416 }
1417}
1418
1419struct ReadPathFilters<'a> {
1421 name: Option<&'a str>,
1423 service: Option<&'a str>,
1425 since: Option<&'a str>,
1427 user: Option<&'a str>,
1429 trace: Option<&'a str>,
1431 level: Option<&'a str>,
1433 search: Option<&'a str>,
1435 project: Option<&'a str>,
1437 release: Option<&'a str>,
1439 environment: Option<&'a str>,
1441 status: Option<&'a str>,
1443 limit: Option<&'a str>,
1445 min_duration_ms: Option<&'a str>,
1447 pagination: Option<&'a str>,
1449 cursor_time: Option<&'a str>,
1451 cursor_id: Option<&'a str>,
1453}
1454
1455fn read_path(target: &ReadTarget, filters: &ReadPathFilters<'_>) -> String {
1457 match target {
1458 ReadTarget::Logs => path_with_query(
1459 "/api/logs",
1460 &[
1461 ("service_name", filters.service),
1462 ("severity", filters.level),
1463 ("search", filters.search),
1464 ("since", filters.since),
1465 ("trace_id", filters.trace),
1466 ("project_id", filters.project),
1467 ("release", filters.release),
1468 ("environment", filters.environment),
1469 ("pagination", filters.pagination),
1470 ("cursor_time", filters.cursor_time),
1471 ("cursor_id", filters.cursor_id),
1472 ("limit", filters.limit),
1473 ],
1474 ),
1475 ReadTarget::Issues => path_with_query(
1476 "/api/telemetry/issues",
1477 &[
1478 ("service_name", filters.service),
1479 ("since", filters.since),
1480 ("status", filters.status),
1481 ("project_id", filters.project),
1482 ("release", filters.release),
1483 ("environment", filters.environment),
1484 ("pagination", filters.pagination),
1485 ("cursor_time", filters.cursor_time),
1486 ("cursor_id", filters.cursor_id),
1487 ("limit", filters.limit),
1488 ],
1489 ),
1490 ReadTarget::Actions => path_with_query(
1491 "/api/telemetry/actions",
1492 &[
1493 ("service_name", filters.service),
1494 ("name", filters.name),
1495 ("since", filters.since),
1496 ("distinct_id", filters.user),
1497 ("project_id", filters.project),
1498 ("release", filters.release),
1499 ("environment", filters.environment),
1500 ("pagination", filters.pagination),
1501 ("cursor_time", filters.cursor_time),
1502 ("cursor_id", filters.cursor_id),
1503 ("limit", filters.limit),
1504 ],
1505 ),
1506 ReadTarget::Releases => path_with_query(
1507 "/api/telemetry/releases",
1508 &[
1509 ("service_name", filters.service),
1510 ("since", filters.since),
1511 ("project_id", filters.project),
1512 ("release", filters.release),
1513 ("environment", filters.environment),
1514 ("limit", filters.limit),
1515 ],
1516 ),
1517 ReadTarget::Traces => path_with_query(
1518 "/api/telemetry/traces",
1519 &[
1520 ("project_id", filters.project),
1521 ("service_name", filters.service),
1522 ("release", filters.release),
1523 ("environment", filters.environment),
1524 ("status", filters.status),
1525 ("since", filters.since),
1526 ("min_duration_ms", filters.min_duration_ms),
1527 ("limit", filters.limit),
1528 ],
1529 ),
1530 ReadTarget::Trace(id) => path_with_query(
1531 &format!("/api/telemetry/traces/{}", encode_component(id)),
1532 &[
1533 ("project_id", filters.project),
1534 ("release", filters.release),
1535 ("environment", filters.environment),
1536 ],
1537 ),
1538 ReadTarget::Issue(id) => format!("/api/telemetry/issues/{}", encode_component(id)),
1539 }
1540}
1541
1542fn explain_path(target: &ExplainTarget) -> String {
1544 match target {
1545 ExplainTarget::Issue(id) => format!("/api/telemetry/issues/{}", encode_component(id)),
1546 ExplainTarget::Trace(id) => format!("/api/telemetry/traces/{}", encode_component(id)),
1547 }
1548}
1549
1550fn set_path(target: &SetTarget) -> String {
1552 match target {
1553 SetTarget::IssueStatus { id, .. } => {
1554 format!("/api/telemetry/issues/{}", encode_component(id))
1555 }
1556 }
1557}
1558
1559fn path_with_query(path: &str, params: &[(&str, Option<&str>)]) -> String {
1561 let query = params
1562 .iter()
1563 .filter_map(|(name, value)| value.map(|v| format!("{name}={}", encode_component(v))))
1564 .collect::<Vec<_>>();
1565
1566 if query.is_empty() {
1567 path.to_owned()
1568 } else {
1569 format!("{path}?{}", query.join("&"))
1570 }
1571}
1572
1573fn encode_component(value: &str) -> String {
1575 let mut encoded = String::new();
1576 for byte in value.bytes() {
1577 if byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.' | b'~') {
1578 encoded.push(char::from(byte));
1579 } else {
1580 encoded.push('%');
1581 encoded.push(hex_digit(byte >> 4));
1582 encoded.push(hex_digit(byte & 0x0f));
1583 }
1584 }
1585 encoded
1586}
1587
1588fn hex_digit(nibble: u8) -> char {
1590 match nibble {
1591 0..=9 => char::from(b'0' + nibble),
1592 10..=15 => char::from(b'A' + (nibble - 10)),
1593 _ => '?',
1594 }
1595}