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::{
38 AuthCredential, execute_login, execute_logout, send_authenticated_with_refresh,
39 token_is_project_ingest_key,
40};
41pub use error::{CliError, RuntimeError, write_cli_error, write_runtime_error};
42use futures_util::StreamExt as _;
43pub use parser::parse_command;
44use render::write_api_success;
45use setup::write_setup_plan;
46use status::execute_status;
47use tokio_tungstenite::connect_async;
48use tokio_tungstenite::tungstenite::{Error as WebSocketError, Message};
49use version::execute_version;
50
51const WATCH_RECONNECT_INITIAL_DELAY: std::time::Duration = std::time::Duration::from_secs(1);
53const WATCH_RECONNECT_MAX_DELAY: std::time::Duration = std::time::Duration::from_secs(30);
55const WATCH_RECONNECT_JITTER_MAX_MILLIS: u64 = 250;
57
58pub(crate) const ISSUE_STATUS_VALUES_NEXT_STEP: &str =
60 "use one of unresolved/open, resolved/closed, ignored";
61pub(crate) const ISSUE_STATUS_FILTER_NEXT_STEP: &str =
63 "use --status unresolved/open, --status resolved/closed, or --status ignored";
64pub(crate) const ISSUE_STATUS_ARGUMENT_NEXT_STEP: &str =
66 "provide one of unresolved/open, resolved/closed, ignored";
67
68#[derive(Debug, Clone, PartialEq, Eq)]
70pub enum Command {
71 Help {
73 topic: HelpTopic,
75 json: bool,
77 },
78 Login {
80 open_browser: bool,
82 json: bool,
84 },
85 Logout {
87 json: bool,
89 },
90 Setup {
92 auto: bool,
94 yes: bool,
96 json: bool,
98 },
99 Status {
101 json: bool,
103 },
104 Doctor {
106 project_id: String,
108 json: bool,
110 },
111 ProjectCreate {
113 options: ProjectCreateOptions,
115 json: bool,
117 },
118 Usage {
120 json: bool,
122 },
123 Version {
125 json: bool,
127 },
128 Read {
130 target: ReadTarget,
132 options: Box<ReadOptions>,
134 json: bool,
136 },
137 Watch {
139 target: WatchTarget,
141 options: WatchOptions,
143 json: bool,
145 },
146 Explain {
148 target: ExplainTarget,
150 json: bool,
152 },
153 InvestigateIssue {
155 issue_id: String,
157 json: bool,
159 },
160 NativeDebugArtifacts {
162 target: NativeDebugArtifactsTarget,
164 json: bool,
166 },
167 Set {
169 target: SetTarget,
171 json: bool,
173 },
174 ProjectSetupSeen {
176 project_id: String,
178 options: ProjectSetupSeenOptions,
180 json: bool,
182 },
183 Support {
185 target: SupportTarget,
187 json: bool,
189 },
190}
191
192#[derive(Debug, Clone, Copy, PartialEq, Eq)]
194pub enum HelpTopic {
195 Root,
197 Login,
199 Logout,
201 Setup,
203 Status,
205 Version,
207 Auth,
209 Json,
211 Examples,
213 Projects,
215 Usage,
217 Read,
219 ReadLogs,
221 ReadIssues,
223 ReadActions,
225 ReadReleases,
227 ReadTraces,
229 ReadTrace,
231 ReadIssue,
233 Watch,
235 Explain,
237 Investigate,
239 NativeDebugArtifacts,
241 Set,
243 Support,
245}
246
247impl HelpTopic {
248 #[must_use]
250 pub const fn key(self) -> &'static str {
251 match self {
252 Self::Root => "root",
253 Self::Login => "login",
254 Self::Logout => "logout",
255 Self::Setup => "setup",
256 Self::Status => "status",
257 Self::Version => "version",
258 Self::Auth => "auth",
259 Self::Json => "json",
260 Self::Examples => "examples",
261 Self::Projects => "projects",
262 Self::Usage => "usage",
263 Self::Read => "read",
264 Self::ReadLogs => "read_logs",
265 Self::ReadIssues => "read_issues",
266 Self::ReadActions => "read_actions",
267 Self::ReadReleases => "read_releases",
268 Self::ReadTraces => "read_traces",
269 Self::ReadTrace => "read_trace",
270 Self::ReadIssue => "read_issue",
271 Self::Watch => "watch",
272 Self::Explain => "explain",
273 Self::Investigate => "investigate",
274 Self::NativeDebugArtifacts => "debug_artifacts",
275 Self::Set => "set",
276 Self::Support => "support",
277 }
278 }
279}
280
281#[derive(Debug, Clone, PartialEq, Eq)]
283pub enum ReadTarget {
284 Logs,
286 Issues,
288 Actions,
290 Releases,
292 Traces,
294 Trace(String),
296 Issue(String),
298}
299
300#[derive(Debug, Clone, Default, PartialEq, Eq)]
302pub struct ReadOptions {
303 pub name: Option<String>,
305 pub service: Option<String>,
307 pub since: Option<String>,
309 pub user: Option<String>,
311 pub trace: Option<String>,
313 pub level: Option<String>,
315 pub search: Option<String>,
317 pub project: Option<String>,
319 pub release: Option<String>,
321 pub environment: Option<String>,
323 pub status: Option<String>,
325 pub limit: Option<String>,
327 pub min_duration_ms: Option<String>,
329 pub pagination: Option<String>,
331 pub cursor_time: Option<String>,
333 pub cursor_id: Option<String>,
335}
336
337impl ReadOptions {
338 #[must_use]
340 pub(crate) fn first_trace_detail_unsupported_flag(&self) -> Option<&'static str> {
341 first_present_flag([
342 (self.name.is_some(), "--name"),
343 (self.service.is_some(), "--service"),
344 (self.since.is_some(), "--since"),
345 (self.user.is_some(), "--user"),
346 (self.trace.is_some(), "--trace"),
347 (self.level.is_some(), "--severity"),
348 (self.search.is_some(), "--search"),
349 (self.status.is_some(), "--status"),
350 (self.limit.is_some(), "--limit"),
351 (self.min_duration_ms.is_some(), "--min-duration-ms"),
352 (self.pagination.is_some(), "--pagination"),
353 (self.cursor_time.is_some(), "--cursor-time"),
354 (self.cursor_id.is_some(), "--cursor-id"),
355 ])
356 }
357
358 #[must_use]
360 pub(crate) fn first_issue_detail_unsupported_flag(&self) -> Option<&'static str> {
361 first_present_flag([
362 (self.name.is_some(), "--name"),
363 (self.service.is_some(), "--service"),
364 (self.since.is_some(), "--since"),
365 (self.user.is_some(), "--user"),
366 (self.trace.is_some(), "--trace"),
367 (self.level.is_some(), "--severity"),
368 (self.search.is_some(), "--search"),
369 (self.project.is_some(), "--project"),
370 (self.release.is_some(), "--release"),
371 (self.environment.is_some(), "--environment"),
372 (self.status.is_some(), "--status"),
373 (self.limit.is_some(), "--limit"),
374 (self.min_duration_ms.is_some(), "--min-duration-ms"),
375 (self.pagination.is_some(), "--pagination"),
376 (self.cursor_time.is_some(), "--cursor-time"),
377 (self.cursor_id.is_some(), "--cursor-id"),
378 ])
379 }
380
381 #[must_use]
383 pub(crate) fn first_log_unsupported_flag(&self) -> Option<&'static str> {
384 first_present_flag([
385 (self.name.is_some(), "--name"),
386 (self.user.is_some(), "--user"),
387 (self.status.is_some(), "--status"),
388 (self.min_duration_ms.is_some(), "--min-duration-ms"),
389 ])
390 }
391
392 #[must_use]
394 pub(crate) fn first_issue_list_unsupported_flag(&self) -> Option<&'static str> {
395 first_present_flag([
396 (self.name.is_some(), "--name"),
397 (self.user.is_some(), "--user"),
398 (self.trace.is_some(), "--trace"),
399 (self.level.is_some(), "--severity"),
400 (self.search.is_some(), "--search"),
401 (self.min_duration_ms.is_some(), "--min-duration-ms"),
402 ])
403 }
404
405 #[must_use]
407 pub(crate) fn first_action_unsupported_flag(&self) -> Option<&'static str> {
408 first_present_flag([
409 (self.trace.is_some(), "--trace"),
410 (self.level.is_some(), "--severity"),
411 (self.search.is_some(), "--search"),
412 (self.status.is_some(), "--status"),
413 (self.min_duration_ms.is_some(), "--min-duration-ms"),
414 ])
415 }
416
417 #[must_use]
419 pub(crate) fn first_release_unsupported_flag(&self) -> Option<&'static str> {
420 first_present_flag([
421 (self.name.is_some(), "--name"),
422 (self.user.is_some(), "--user"),
423 (self.trace.is_some(), "--trace"),
424 (self.level.is_some(), "--severity"),
425 (self.search.is_some(), "--search"),
426 (self.status.is_some(), "--status"),
427 (self.min_duration_ms.is_some(), "--min-duration-ms"),
428 (self.pagination.is_some(), "--pagination"),
429 (self.cursor_time.is_some(), "--cursor-time"),
430 (self.cursor_id.is_some(), "--cursor-id"),
431 ])
432 }
433
434 #[must_use]
436 pub(crate) fn first_trace_list_unsupported_flag(&self) -> Option<&'static str> {
437 first_present_flag([
438 (self.name.is_some(), "--name"),
439 (self.user.is_some(), "--user"),
440 (self.trace.is_some(), "--trace"),
441 (self.level.is_some(), "--severity"),
442 (self.search.is_some(), "--search"),
443 (self.pagination.is_some(), "--pagination"),
444 (self.cursor_time.is_some(), "--cursor-time"),
445 (self.cursor_id.is_some(), "--cursor-id"),
446 ])
447 }
448}
449
450fn first_present_flag<const N: usize>(flags: [(bool, &'static str); N]) -> Option<&'static str> {
452 flags
453 .iter()
454 .find_map(|(present, flag)| present.then_some(*flag))
455}
456
457#[derive(Debug, Clone, Copy, PartialEq, Eq)]
459pub enum WatchTarget {
460 All,
462 Logs,
464 Issues,
466 Actions,
468}
469
470#[derive(Debug, Clone, Default, PartialEq, Eq)]
472pub struct WatchOptions {
473 pub severity: Vec<String>,
475}
476
477#[derive(Debug, Clone, Default, PartialEq, Eq)]
479pub struct ProjectSetupSeenOptions {
480 pub runtime: Option<String>,
482 pub source: Option<String>,
484 pub environment: Option<String>,
486}
487
488#[derive(Debug, Clone, PartialEq, Eq)]
490pub struct ProjectCreateOptions {
491 pub name: String,
493 pub runtime: Option<String>,
495 pub environment: Option<String>,
497 pub ingest_key_file: String,
499 pub abandon_retry: bool,
501}
502
503#[derive(Debug, Clone, PartialEq, Eq)]
505pub enum NativeDebugArtifactsTarget {
506 Upload(NativeDebugUploadOptions),
508 Lookup(NativeDebugLookupOptions),
510}
511
512#[derive(Debug, Clone, PartialEq, Eq)]
514pub struct NativeDebugLookupOptions {
515 pub project_id: String,
517 pub release: String,
519 pub environment: String,
521 pub service: String,
523 pub image_uuid: String,
525 pub architecture: String,
527}
528
529#[derive(Debug, Clone, PartialEq, Eq)]
531pub struct NativeDebugUploadOptions {
532 pub path: String,
534 pub project_id: String,
536 pub release: String,
538 pub environment: String,
540 pub service: String,
542 pub expected_image_uuids: Vec<String>,
544 pub dry_run: bool,
546}
547
548#[derive(Debug, Clone, PartialEq, Eq)]
550pub enum SupportTarget {
551 Create(Box<SupportTicketCreateOptions>),
553 List(Box<SupportTicketListOptions>),
555 Detail(String),
557 ContextHistory {
559 ticket_id: String,
561 },
562 ReplyContext(Box<SupportContextReplyOptions>),
564 UpdateStatus {
566 ticket_id: String,
568 status: SupportTicketLifecycleStatus,
570 },
571}
572
573#[derive(Debug, Clone, Default, PartialEq, Eq)]
575pub struct SupportContextReplyOptions {
576 pub ticket_id: String,
578 pub context: String,
580 pub retry_key: String,
582 pub diagnostics: bool,
584}
585
586#[derive(Debug, Clone, Copy, PartialEq, Eq)]
588pub enum SupportTicketLifecycleStatus {
589 Open,
591 Closed,
593}
594
595impl SupportTicketLifecycleStatus {
596 #[must_use]
598 pub const fn as_str(self) -> &'static str {
599 match self {
600 Self::Open => "open",
601 Self::Closed => "closed",
602 }
603 }
604}
605
606#[derive(Debug, Clone, Default, PartialEq, Eq)]
608pub struct SupportTicketCreateOptions {
609 pub category: String,
611 pub title: String,
613 pub description: String,
615 pub project_id: Option<String>,
617 pub environment: Option<String>,
619 pub runtime: Option<String>,
621 pub framework: Option<String>,
623 pub sdk_package: Option<String>,
625 pub sdk_version: Option<String>,
627 pub release: Option<String>,
629 pub trace_id: Option<String>,
631 pub event_id: Option<String>,
633 pub diagnostics: bool,
635}
636
637#[derive(Debug, Clone, Default, PartialEq, Eq)]
639pub struct SupportTicketListOptions {
640 pub project_id: Option<String>,
642 pub status: Option<String>,
644 pub source: Option<String>,
646 pub category: Option<String>,
648 pub release: Option<String>,
650 pub limit: Option<String>,
652 pub pagination: Option<String>,
654 pub cursor_time: Option<String>,
656 pub cursor_id: Option<String>,
658}
659
660#[derive(Debug, Clone, PartialEq, Eq)]
662pub enum ExplainTarget {
663 Issue(String),
665 Trace(String),
667}
668
669#[derive(Debug, Clone, PartialEq, Eq)]
671pub enum SetTarget {
672 IssueStatus {
674 id: String,
676 status: String,
678 },
679}
680
681#[derive(Debug, Clone, PartialEq, Eq)]
683pub struct CliEnvironment {
684 pub base_url: String,
686 pub token: Option<String>,
688 pub home: Option<std::path::PathBuf>,
690 pub cwd: Option<std::path::PathBuf>,
692}
693
694impl CliEnvironment {
695 #[must_use]
697 pub fn from_process() -> Self {
698 Self {
699 base_url: std::env::var("LOGBREW_API_URL")
700 .unwrap_or_else(|_| String::from("https://api.logbrew.co")),
701 token: std::env::var("LOGBREW_TOKEN").ok(),
702 home: std::env::var_os("HOME").map(std::path::PathBuf::from),
703 cwd: std::env::current_dir().ok(),
704 }
705 }
706}
707
708impl Command {
709 #[must_use]
711 pub fn http_path(&self) -> Option<String> {
712 match self {
713 Self::Read {
714 target, options, ..
715 } => Some(read_path(
716 target,
717 &ReadPathFilters {
718 name: options.name.as_deref(),
719 service: options.service.as_deref(),
720 since: options.since.as_deref(),
721 user: options.user.as_deref(),
722 trace: options.trace.as_deref(),
723 level: options.level.as_deref(),
724 search: options.search.as_deref(),
725 project: options.project.as_deref(),
726 release: options.release.as_deref(),
727 environment: options.environment.as_deref(),
728 status: options.status.as_deref(),
729 limit: options.limit.as_deref(),
730 min_duration_ms: options.min_duration_ms.as_deref(),
731 pagination: options.pagination.as_deref(),
732 cursor_time: options.cursor_time.as_deref(),
733 cursor_id: options.cursor_id.as_deref(),
734 },
735 )),
736 Self::Explain { target, .. } => Some(explain_path(target)),
737 Self::Set { target, .. } => Some(set_path(target)),
738 Self::ProjectSetupSeen { project_id, .. } => {
739 Some(format!("/api/projects/{project_id}/setup/seen"))
740 }
741 Self::ProjectCreate { .. } => Some(String::from("/api/projects")),
742 Self::Support { target, .. } => Some(support::path(target)),
743 Self::Help { .. }
744 | Self::Login { .. }
745 | Self::Logout { .. }
746 | Self::Setup { .. }
747 | Self::Status { .. }
748 | Self::Doctor { .. }
749 | Self::Usage { .. }
750 | Self::Version { .. }
751 | Self::InvestigateIssue { .. }
752 | Self::NativeDebugArtifacts { .. }
753 | Self::Watch { .. } => None,
754 }
755 }
756
757 #[must_use]
759 pub const fn wants_json(&self) -> bool {
760 match self {
761 Self::Help { json, .. }
762 | Self::Login { json, .. }
763 | Self::Logout { json }
764 | Self::Status { json }
765 | Self::Doctor { json, .. }
766 | Self::ProjectCreate { json, .. }
767 | Self::Usage { json }
768 | Self::Version { json }
769 | Self::Read { json, .. }
770 | Self::Watch { json, .. }
771 | Self::Explain { json, .. }
772 | Self::InvestigateIssue { json, .. }
773 | Self::NativeDebugArtifacts { json, .. }
774 | Self::Set { json, .. }
775 | Self::ProjectSetupSeen { json, .. }
776 | Self::Support { json, .. }
777 | Self::Setup { json, .. } => *json,
778 }
779 }
780
781 #[must_use]
783 pub const fn http_method(&self) -> Option<HttpMethod> {
784 match self {
785 Self::ProjectCreate { .. }
786 | Self::ProjectSetupSeen { .. }
787 | Self::Support {
788 target: SupportTarget::Create(_),
789 ..
790 }
791 | Self::Support {
792 target: SupportTarget::ReplyContext(_),
793 ..
794 } => Some(HttpMethod::Post),
795 Self::Support {
796 target: SupportTarget::UpdateStatus { .. },
797 ..
798 }
799 | Self::Set { .. } => Some(HttpMethod::Patch),
800 Self::Read { .. } | Self::Explain { .. } | Self::Support { .. } => {
801 Some(HttpMethod::Get)
802 }
803 Self::Help { .. }
804 | Self::Login { .. }
805 | Self::Logout { .. }
806 | Self::Setup { .. }
807 | Self::Status { .. }
808 | Self::Doctor { .. }
809 | Self::Usage { .. }
810 | Self::Version { .. }
811 | Self::InvestigateIssue { .. }
812 | Self::NativeDebugArtifacts { .. }
813 | Self::Watch { .. } => None,
814 }
815 }
816
817 #[must_use]
819 pub fn request_body(&self) -> Option<serde_json::Value> {
820 self.request_body_for_token(None)
821 }
822
823 #[must_use]
825 fn request_body_for_token(&self, token: Option<&str>) -> Option<serde_json::Value> {
826 match self {
827 Self::Set {
828 target: SetTarget::IssueStatus { status, .. },
829 ..
830 } => Some(serde_json::json!({ "status": status })),
831 Self::ProjectSetupSeen { options, .. } => Some(project_setup_seen_body(options, token)),
832 Self::ProjectCreate { options, .. } => Some(project_create_body(options)),
833 Self::Support {
834 target: SupportTarget::Create(options),
835 ..
836 } => Some(support::create_body(options)),
837 Self::Support {
838 target: SupportTarget::ReplyContext(options),
839 ..
840 } => Some(support::context_body(options)),
841 Self::Support {
842 target: SupportTarget::UpdateStatus { status, .. },
843 ..
844 } => Some(serde_json::json!({"status": status.as_str()})),
845 Self::Help { .. }
846 | Self::Login { .. }
847 | Self::Logout { .. }
848 | Self::Setup { .. }
849 | Self::Status { .. }
850 | Self::Doctor { .. }
851 | Self::Usage { .. }
852 | Self::Version { .. }
853 | Self::Read { .. }
854 | Self::Watch { .. }
855 | Self::Explain { .. }
856 | Self::InvestigateIssue { .. }
857 | Self::NativeDebugArtifacts { .. }
858 | Self::Support { .. } => None,
859 }
860 }
861
862 fn idempotency_key(&self) -> Option<&str> {
864 match self {
865 Self::Support {
866 target: SupportTarget::ReplyContext(options),
867 ..
868 } => Some(options.retry_key.as_str()),
869 Self::Help { .. }
870 | Self::Login { .. }
871 | Self::Logout { .. }
872 | Self::Setup { .. }
873 | Self::Status { .. }
874 | Self::Doctor { .. }
875 | Self::Usage { .. }
876 | Self::Version { .. }
877 | Self::Read { .. }
878 | Self::Watch { .. }
879 | Self::Explain { .. }
880 | Self::InvestigateIssue { .. }
881 | Self::NativeDebugArtifacts { .. }
882 | Self::Set { .. }
883 | Self::ProjectSetupSeen { .. }
884 | Self::ProjectCreate { .. }
885 | Self::Support { .. } => None,
886 }
887 }
888}
889
890#[derive(Debug, Clone, Copy, PartialEq, Eq)]
892pub enum HttpMethod {
893 Get,
895 Post,
897 Patch,
899}
900
901fn project_setup_seen_body(
903 options: &ProjectSetupSeenOptions,
904 token: Option<&str>,
905) -> serde_json::Value {
906 let mut body = serde_json::Map::new();
907 if let Some(runtime) = options.runtime.as_ref() {
908 drop(body.insert(
909 "runtime".to_owned(),
910 serde_json::Value::String(runtime.clone()),
911 ));
912 }
913 if let Some(source) = setup_seen_source(options, token) {
914 drop(body.insert("source".to_owned(), serde_json::Value::String(source)));
915 }
916 if let Some(environment) = options.environment.as_ref() {
917 drop(body.insert(
918 "environment".to_owned(),
919 serde_json::Value::String(environment.clone()),
920 ));
921 }
922 serde_json::Value::Object(body)
923}
924
925fn project_create_body(options: &ProjectCreateOptions) -> serde_json::Value {
927 let mut body = serde_json::Map::new();
928 drop(body.insert(
929 "name".to_owned(),
930 serde_json::Value::String(options.name.clone()),
931 ));
932 if let Some(runtime) = options.runtime.as_ref() {
933 drop(body.insert(
934 "runtime".to_owned(),
935 serde_json::Value::String(runtime.clone()),
936 ));
937 }
938 if let Some(environment) = options.environment.as_ref() {
939 drop(body.insert(
940 "environment".to_owned(),
941 serde_json::Value::String(environment.clone()),
942 ));
943 }
944 drop(body.insert(
945 "source".to_owned(),
946 serde_json::Value::String(String::from("cli")),
947 ));
948 serde_json::Value::Object(body)
949}
950
951fn setup_seen_source(options: &ProjectSetupSeenOptions, token: Option<&str>) -> Option<String> {
953 if token_is_project_ingest_key(token) {
954 return None;
955 }
956 Some(options.source.as_deref().unwrap_or("cli").to_owned())
957}
958
959pub async fn execute_command<W: std::io::Write>(
965 command: &Command,
966 env: &CliEnvironment,
967 output: &mut W,
968) -> Result<(), RuntimeError> {
969 match command {
970 Command::Help { topic, json } => execute_help(*topic, *json, output),
971 Command::Login { open_browser, json } => {
972 execute_login(env, *open_browser, *json, output).await
973 }
974 Command::Logout { json } => execute_logout(env, *json, output).await,
975 Command::Setup { auto, yes, json } => execute_setup(env, *auto, *yes, *json, output),
976 Command::Status { json } => execute_status(env, *json, output).await,
977 Command::Doctor { project_id, json } => {
978 doctor::execute(env, project_id.as_str(), *json, output).await
979 }
980 Command::ProjectCreate { options, json } => {
981 project_create::execute(env, options, *json, output).await
982 }
983 Command::Usage { json } => usage::execute(env, *json, output).await,
984 Command::Version { json } => execute_version(*json, output),
985 Command::InvestigateIssue { issue_id, json } => {
986 investigate::execute(env, issue_id.as_str(), *json, output).await
987 }
988 Command::NativeDebugArtifacts { target, json } => {
989 native_debug_artifacts::execute(env, target, *json, output).await
990 }
991 Command::Read { .. }
992 | Command::Explain { .. }
993 | Command::Set { .. }
994 | Command::ProjectSetupSeen { .. }
995 | Command::Support { .. } => execute_http(command, env, output).await,
996 Command::Watch {
997 target,
998 options,
999 json,
1000 } => execute_watch(env, *target, options, *json, output).await,
1001 }
1002}
1003
1004fn execute_help<W: std::io::Write>(
1006 topic: HelpTopic,
1007 json: bool,
1008 output: &mut W,
1009) -> Result<(), RuntimeError> {
1010 let help = help::help_text(topic);
1011 if json {
1012 let body = serde_json::json!({
1013 "ok": true,
1014 "topic": topic.key(),
1015 "help": help,
1016 });
1017 writeln!(output, "{body}")?;
1018 } else {
1019 writeln!(output, "{help}")?;
1020 }
1021 Ok(())
1022}
1023
1024fn execute_setup<W: std::io::Write>(
1026 env: &CliEnvironment,
1027 auto: bool,
1028 yes: bool,
1029 json: bool,
1030 output: &mut W,
1031) -> Result<(), RuntimeError> {
1032 write_setup_plan(env.cwd.as_deref(), auto, yes, json, output)?;
1033 Ok(())
1034}
1035
1036async fn execute_http<W: std::io::Write>(
1038 command: &Command,
1039 env: &CliEnvironment,
1040 output: &mut W,
1041) -> Result<(), RuntimeError> {
1042 let path = command.http_path().ok_or(CliError::UnknownCommand)?;
1043 let url = format!("{}{}", env.base_url.trim_end_matches('/'), path);
1044 let client = reqwest::Client::builder()
1045 .timeout(std::time::Duration::from_secs(30))
1046 .connect_timeout(std::time::Duration::from_secs(10))
1047 .build()?;
1048
1049 let support_command = matches!(command, Command::Support { .. });
1050 let response_result = send_authenticated_with_refresh(&client, env, |client, credential| {
1051 build_command_request(client, command, url.as_str(), credential)
1052 })
1053 .await;
1054 let (response, credential) = match response_result {
1055 Ok(response) => response,
1056 Err(RuntimeError::Http(_)) if support_command => return Err(support_transport_error()),
1057 Err(error) => return Err(error),
1058 };
1059 let status = response.status();
1060 let body = match response.text().await {
1061 Ok(body) => body,
1062 Err(_) if support_command => return Err(support_transport_error()),
1063 Err(error) => return Err(RuntimeError::Http(error)),
1064 };
1065
1066 if !status.is_success() {
1067 let body = if let Command::Support { target, .. } = command {
1068 support::safe_error_body(target, status.as_u16())
1069 } else {
1070 credential.redact_response_body(body.as_str())
1071 };
1072 return Err(RuntimeError::Api {
1073 status: status.as_u16(),
1074 body,
1075 auth_source: credential.source(),
1076 auth_label: credential.label(),
1077 });
1078 }
1079
1080 write_api_success(command, body.as_str(), output)?;
1081 Ok(())
1082}
1083
1084const fn support_transport_error() -> RuntimeError {
1086 RuntimeError::Unavailable {
1087 message: "support request could not be completed",
1088 next: "check network connectivity and retry the support command",
1089 }
1090}
1091
1092fn build_command_request(
1094 client: &reqwest::Client,
1095 command: &Command,
1096 url: &str,
1097 credential: &AuthCredential,
1098) -> reqwest::RequestBuilder {
1099 let mut request = match command.http_method().unwrap_or(HttpMethod::Get) {
1100 HttpMethod::Get => client.get(url),
1101 HttpMethod::Post => client.post(url),
1102 HttpMethod::Patch => client.patch(url),
1103 }
1104 .bearer_auth(credential.token());
1105 if let Some(body) = command.request_body_for_token(Some(credential.token())) {
1106 request = request.json(&body);
1107 }
1108 if let Some(key) = command.idempotency_key() {
1109 request = request.header("Idempotency-Key", key);
1110 }
1111 request
1112}
1113
1114async fn execute_watch<W: std::io::Write>(
1116 env: &CliEnvironment,
1117 target: WatchTarget,
1118 options: &WatchOptions,
1119 json: bool,
1120 output: &mut W,
1121) -> Result<(), RuntimeError> {
1122 if !json {
1123 return Err(RuntimeError::Unavailable {
1124 message: "watch streams JSON for agents",
1125 next: "run logbrew watch --json",
1126 });
1127 }
1128
1129 let mut reconnect_backoff = WatchReconnectBackoff::default();
1130 loop {
1131 let ticket = match request_feed_ticket(env).await {
1132 Ok(ticket) => ticket,
1133 Err(error) if reconnect_backoff.connected_once() && !runtime_error_is_auth(&error) => {
1134 tokio::time::sleep(reconnect_backoff.next_delay()).await;
1135 continue;
1136 }
1137 Err(error) => return Err(error),
1138 };
1139 let live_url = feed_live_url(env.base_url.as_str(), ticket.as_str())?;
1140 let (mut websocket, _) = match connect_async(live_url.as_str()).await {
1141 Ok(connection) => connection,
1142 Err(error)
1143 if reconnect_backoff.connected_once() && !websocket_error_is_auth(&error) =>
1144 {
1145 tokio::time::sleep(reconnect_backoff.next_delay()).await;
1146 continue;
1147 }
1148 Err(error) => return Err(map_websocket_connect_error(error)),
1149 };
1150 reconnect_backoff.mark_connected();
1151
1152 let mut emitted_before_disconnect = false;
1153 loop {
1154 let Some(message) = websocket.next().await else {
1155 break;
1156 };
1157 let message = match message {
1158 Ok(message) => message,
1159 Err(error) if websocket_error_is_auth(&error) => {
1160 return Err(map_websocket_stream_error(error));
1161 }
1162 Err(_) => break,
1163 };
1164 match message {
1165 Message::Text(text) => {
1166 let event = parse_live_event(text.as_str())?;
1167 if watch_event_matches(target, options, &event) {
1168 writeln!(output, "{event}")?;
1169 }
1170 emitted_before_disconnect = true;
1171 }
1172 Message::Binary(_) | Message::Ping(_) | Message::Pong(_) | Message::Frame(_) => {}
1173 Message::Close(_) => return Ok(()),
1174 }
1175 }
1176 if emitted_before_disconnect {
1177 reconnect_backoff.reset();
1178 }
1179 tokio::time::sleep(reconnect_backoff.next_delay()).await;
1180 }
1181}
1182
1183#[derive(Debug, Default)]
1185struct WatchReconnectBackoff {
1186 connected_once: bool,
1188 attempts: u32,
1190}
1191
1192impl WatchReconnectBackoff {
1193 const fn connected_once(&self) -> bool {
1195 self.connected_once
1196 }
1197
1198 const fn mark_connected(&mut self) {
1200 self.connected_once = true;
1201 }
1202
1203 const fn reset(&mut self) {
1205 self.attempts = 0;
1206 }
1207
1208 fn next_delay(&mut self) -> std::time::Duration {
1210 let exponent = self.attempts.min(5);
1211 let multiplier = 1_u64 << exponent;
1212 self.attempts = self.attempts.saturating_add(1);
1213 let base = WATCH_RECONNECT_INITIAL_DELAY
1214 .as_secs()
1215 .saturating_mul(multiplier)
1216 .min(WATCH_RECONNECT_MAX_DELAY.as_secs());
1217 let delay = std::time::Duration::from_secs(base) + watch_reconnect_jitter();
1218 if delay > WATCH_RECONNECT_MAX_DELAY {
1219 WATCH_RECONNECT_MAX_DELAY
1220 } else {
1221 delay
1222 }
1223 }
1224}
1225
1226fn watch_reconnect_jitter() -> std::time::Duration {
1228 let Ok(elapsed) = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH) else {
1229 return std::time::Duration::ZERO;
1230 };
1231 std::time::Duration::from_millis(
1232 u64::from(elapsed.subsec_millis()) % WATCH_RECONNECT_JITTER_MAX_MILLIS,
1233 )
1234}
1235
1236const fn runtime_error_is_auth(error: &RuntimeError) -> bool {
1238 matches!(
1239 error,
1240 RuntimeError::MissingToken | RuntimeError::Api { status: 401, .. }
1241 )
1242}
1243
1244fn websocket_error_is_auth(error: &WebSocketError) -> bool {
1246 matches!(error, WebSocketError::Http(response) if response.status().as_u16() == 401)
1247}
1248
1249async fn request_feed_ticket(env: &CliEnvironment) -> Result<String, RuntimeError> {
1251 let url = format!("{}/api/feed/ticket", env.base_url.trim_end_matches('/'));
1252 let client = reqwest::Client::builder()
1253 .timeout(std::time::Duration::from_secs(30))
1254 .connect_timeout(std::time::Duration::from_secs(10))
1255 .build()?;
1256 let (response, credential) =
1257 send_authenticated_with_refresh(&client, env, |client, credential| {
1258 client.post(url.as_str()).bearer_auth(credential.token())
1259 })
1260 .await?;
1261 let status = response.status();
1262 let body = response.text().await?;
1263 if !status.is_success() {
1264 return Err(RuntimeError::Api {
1265 status: status.as_u16(),
1266 body: credential.redact_response_body(body.as_str()),
1267 auth_source: credential.source(),
1268 auth_label: credential.label(),
1269 });
1270 }
1271
1272 let value = serde_json::from_str::<serde_json::Value>(body.as_str()).map_err(|_| {
1273 RuntimeError::Unavailable {
1274 message: "feed ticket response was not valid JSON",
1275 next: "retry logbrew watch or run logbrew status",
1276 }
1277 })?;
1278 value
1279 .get("ticket")
1280 .and_then(serde_json::Value::as_str)
1281 .map(str::trim)
1282 .filter(|ticket| !ticket.is_empty())
1283 .map(ToOwned::to_owned)
1284 .ok_or(RuntimeError::Unavailable {
1285 message: "feed ticket response did not include a ticket",
1286 next: "retry logbrew watch or run logbrew status",
1287 })
1288}
1289
1290fn feed_live_url(base_url: &str, ticket: &str) -> Result<String, RuntimeError> {
1292 let trimmed = base_url.trim_end_matches('/');
1293 let (scheme, rest) = websocket_base_parts(trimmed).ok_or(RuntimeError::Unavailable {
1294 message: "LOGBREW_API_URL must start with http:// or https://",
1295 next: "check LOGBREW_API_URL or run logbrew status",
1296 })?;
1297 Ok(format!(
1298 "{scheme}://{rest}/api/feed/live?ticket={}",
1299 encode_component(ticket)
1300 ))
1301}
1302
1303fn websocket_base_parts(base_url: &str) -> Option<(&'static str, &str)> {
1305 base_url
1306 .strip_prefix("https://")
1307 .map(|rest| ("wss", rest))
1308 .or_else(|| base_url.strip_prefix("http://").map(|rest| ("ws", rest)))
1309}
1310
1311fn parse_live_event(text: &str) -> Result<serde_json::Value, RuntimeError> {
1313 serde_json::from_str::<serde_json::Value>(text).map_err(|_| RuntimeError::Unavailable {
1314 message: "live watch event was not valid JSON",
1315 next: "retry logbrew watch or check LOGBREW_API_URL",
1316 })
1317}
1318
1319fn watch_event_matches(
1321 target: WatchTarget,
1322 options: &WatchOptions,
1323 event: &serde_json::Value,
1324) -> bool {
1325 target_matches_event(target, event) && severity_matches(options, event)
1326}
1327
1328fn target_matches_event(target: WatchTarget, event: &serde_json::Value) -> bool {
1330 let event_type = event
1331 .get("type")
1332 .and_then(serde_json::Value::as_str)
1333 .unwrap_or_default();
1334 match target {
1335 WatchTarget::All => true,
1336 WatchTarget::Logs => event_type == "native_log",
1337 WatchTarget::Issues => event_type == "native_issue",
1338 WatchTarget::Actions => event_type == "native_action",
1339 }
1340}
1341
1342fn severity_matches(options: &WatchOptions, event: &serde_json::Value) -> bool {
1344 if options.severity.is_empty() {
1345 return true;
1346 }
1347 let Some(severity) = event
1348 .get("data")
1349 .and_then(|data| data.get("severity").or_else(|| data.get("level")))
1350 .and_then(serde_json::Value::as_str)
1351 else {
1352 return false;
1353 };
1354 options
1355 .severity
1356 .iter()
1357 .any(|allowed| allowed.as_str() == severity)
1358}
1359
1360fn map_websocket_connect_error(error: WebSocketError) -> RuntimeError {
1362 match error {
1363 WebSocketError::Http(response) if response.status().as_u16() == 401 => {
1364 RuntimeError::Unavailable {
1365 message: "live watch ticket was rejected",
1366 next: "run logbrew login",
1367 }
1368 }
1369 WebSocketError::Http(_) => RuntimeError::Unavailable {
1370 message: "live watch websocket upgrade failed",
1371 next: "retry logbrew watch or check LOGBREW_API_URL",
1372 },
1373 WebSocketError::ConnectionClosed
1374 | WebSocketError::AlreadyClosed
1375 | WebSocketError::Io(_)
1376 | WebSocketError::Tls(_)
1377 | WebSocketError::Capacity(_)
1378 | WebSocketError::Protocol(_)
1379 | WebSocketError::WriteBufferFull(_)
1380 | WebSocketError::Utf8(_)
1381 | WebSocketError::AttackAttempt
1382 | WebSocketError::Url(_)
1383 | WebSocketError::HttpFormat(_) => RuntimeError::Unavailable {
1384 message: "live watch websocket failed",
1385 next: "retry logbrew watch or check LOGBREW_API_URL",
1386 },
1387 }
1388}
1389
1390fn map_websocket_stream_error(error: WebSocketError) -> RuntimeError {
1392 match error {
1393 WebSocketError::ConnectionClosed | WebSocketError::AlreadyClosed => {
1394 RuntimeError::Unavailable {
1395 message: "live watch websocket closed",
1396 next: "retry logbrew watch",
1397 }
1398 }
1399 WebSocketError::Http(response) if response.status().as_u16() == 401 => {
1400 RuntimeError::Unavailable {
1401 message: "live watch ticket was rejected",
1402 next: "run logbrew login",
1403 }
1404 }
1405 WebSocketError::Http(_)
1406 | WebSocketError::Io(_)
1407 | WebSocketError::Tls(_)
1408 | WebSocketError::Capacity(_)
1409 | WebSocketError::Protocol(_)
1410 | WebSocketError::WriteBufferFull(_)
1411 | WebSocketError::Utf8(_)
1412 | WebSocketError::AttackAttempt
1413 | WebSocketError::Url(_)
1414 | WebSocketError::HttpFormat(_) => RuntimeError::Unavailable {
1415 message: "live watch websocket failed",
1416 next: "retry logbrew watch or check LOGBREW_API_URL",
1417 },
1418 }
1419}
1420
1421struct ReadPathFilters<'a> {
1423 name: Option<&'a str>,
1425 service: Option<&'a str>,
1427 since: Option<&'a str>,
1429 user: Option<&'a str>,
1431 trace: Option<&'a str>,
1433 level: Option<&'a str>,
1435 search: Option<&'a str>,
1437 project: Option<&'a str>,
1439 release: Option<&'a str>,
1441 environment: Option<&'a str>,
1443 status: Option<&'a str>,
1445 limit: Option<&'a str>,
1447 min_duration_ms: Option<&'a str>,
1449 pagination: Option<&'a str>,
1451 cursor_time: Option<&'a str>,
1453 cursor_id: Option<&'a str>,
1455}
1456
1457fn read_path(target: &ReadTarget, filters: &ReadPathFilters<'_>) -> String {
1459 match target {
1460 ReadTarget::Logs => path_with_query(
1461 "/api/logs",
1462 &[
1463 ("service_name", filters.service),
1464 ("severity", filters.level),
1465 ("search", filters.search),
1466 ("since", filters.since),
1467 ("trace_id", filters.trace),
1468 ("project_id", filters.project),
1469 ("release", filters.release),
1470 ("environment", filters.environment),
1471 ("pagination", filters.pagination),
1472 ("cursor_time", filters.cursor_time),
1473 ("cursor_id", filters.cursor_id),
1474 ("limit", filters.limit),
1475 ],
1476 ),
1477 ReadTarget::Issues => path_with_query(
1478 "/api/telemetry/issues",
1479 &[
1480 ("service_name", filters.service),
1481 ("since", filters.since),
1482 ("status", filters.status),
1483 ("project_id", filters.project),
1484 ("release", filters.release),
1485 ("environment", filters.environment),
1486 ("pagination", filters.pagination),
1487 ("cursor_time", filters.cursor_time),
1488 ("cursor_id", filters.cursor_id),
1489 ("limit", filters.limit),
1490 ],
1491 ),
1492 ReadTarget::Actions => path_with_query(
1493 "/api/telemetry/actions",
1494 &[
1495 ("service_name", filters.service),
1496 ("name", filters.name),
1497 ("since", filters.since),
1498 ("distinct_id", filters.user),
1499 ("project_id", filters.project),
1500 ("release", filters.release),
1501 ("environment", filters.environment),
1502 ("pagination", filters.pagination),
1503 ("cursor_time", filters.cursor_time),
1504 ("cursor_id", filters.cursor_id),
1505 ("limit", filters.limit),
1506 ],
1507 ),
1508 ReadTarget::Releases => path_with_query(
1509 "/api/telemetry/releases",
1510 &[
1511 ("service_name", filters.service),
1512 ("since", filters.since),
1513 ("project_id", filters.project),
1514 ("release", filters.release),
1515 ("environment", filters.environment),
1516 ("limit", filters.limit),
1517 ],
1518 ),
1519 ReadTarget::Traces => path_with_query(
1520 "/api/telemetry/traces",
1521 &[
1522 ("project_id", filters.project),
1523 ("service_name", filters.service),
1524 ("release", filters.release),
1525 ("environment", filters.environment),
1526 ("status", filters.status),
1527 ("since", filters.since),
1528 ("min_duration_ms", filters.min_duration_ms),
1529 ("limit", filters.limit),
1530 ],
1531 ),
1532 ReadTarget::Trace(id) => path_with_query(
1533 &format!("/api/telemetry/traces/{}", encode_component(id)),
1534 &[
1535 ("project_id", filters.project),
1536 ("release", filters.release),
1537 ("environment", filters.environment),
1538 ],
1539 ),
1540 ReadTarget::Issue(id) => format!("/api/telemetry/issues/{}", encode_component(id)),
1541 }
1542}
1543
1544fn explain_path(target: &ExplainTarget) -> String {
1546 match target {
1547 ExplainTarget::Issue(id) => format!("/api/telemetry/issues/{}", encode_component(id)),
1548 ExplainTarget::Trace(id) => format!("/api/telemetry/traces/{}", encode_component(id)),
1549 }
1550}
1551
1552fn set_path(target: &SetTarget) -> String {
1554 match target {
1555 SetTarget::IssueStatus { id, .. } => {
1556 format!("/api/telemetry/issues/{}", encode_component(id))
1557 }
1558 }
1559}
1560
1561fn path_with_query(path: &str, params: &[(&str, Option<&str>)]) -> String {
1563 let query = params
1564 .iter()
1565 .filter_map(|(name, value)| value.map(|v| format!("{name}={}", encode_component(v))))
1566 .collect::<Vec<_>>();
1567
1568 if query.is_empty() {
1569 path.to_owned()
1570 } else {
1571 format!("{path}?{}", query.join("&"))
1572 }
1573}
1574
1575fn encode_component(value: &str) -> String {
1577 let mut encoded = String::new();
1578 for byte in value.bytes() {
1579 if byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.' | b'~') {
1580 encoded.push(char::from(byte));
1581 } else {
1582 encoded.push('%');
1583 encoded.push(hex_digit(byte >> 4));
1584 encoded.push(hex_digit(byte & 0x0f));
1585 }
1586 }
1587 encoded
1588}
1589
1590fn hex_digit(nibble: u8) -> char {
1592 match nibble {
1593 0..=9 => char::from(b'0' + nibble),
1594 10..=15 => char::from(b'A' + (nibble - 10)),
1595 _ => '?',
1596 }
1597}