1use std::future::{Future, IntoFuture};
2use std::pin::Pin;
3use std::sync::Arc;
4use std::time::{Duration, SystemTime};
5
6use reqwest::header::{ACCEPT, AUTHORIZATION, CONTENT_TYPE, LOCATION, RETRY_AFTER, USER_AGENT};
7use tokio::time::Instant;
8
9use crate::error::{Error, Failure, FieldError, Result};
10use crate::message::{
11 Accepted, Category, Defaults, EventType, Message, MetadataValue, Priority, Severity,
12};
13use crate::validate::{encode_message, valid_idempotency_key};
14
15pub const VERSION: &str = env!("CARGO_PKG_VERSION");
17
18const DEFAULT_TIMEOUT: Duration = Duration::from_secs(5);
19const DEFAULT_RETRIES: u32 = 4;
20const DEFAULT_DEADLINE: Duration = Duration::from_secs(30);
21const DEFAULT_BACKOFF_BASE: Duration = Duration::from_millis(500);
22const DEFAULT_BACKOFF_MAX: Duration = Duration::from_secs(8);
23const MAX_ANSWER_BYTES: usize = 64 << 10;
24
25#[derive(Clone)]
39pub struct Honk {
40 inner: Arc<Inner>,
41}
42
43struct Inner {
44 endpoint: String,
45 key: String,
46 timeout: Duration,
47 retries: u32,
48 deadline: Duration,
49 backoff_base: Duration,
50 backoff_max: Duration,
51 defaults: Defaults,
52 validate: bool,
53 user_agent: String,
54 http: reqwest::Client,
55}
56
57impl std::fmt::Debug for Honk {
58 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
59 f.debug_struct("Honk")
61 .field("endpoint", &self.inner.endpoint)
62 .finish_non_exhaustive()
63 }
64}
65
66#[derive(Debug, Clone)]
84#[must_use]
85pub struct HonkBuilder {
86 url: String,
87 key: String,
88 timeout: Duration,
89 retries: u32,
90 deadline: Duration,
91 backoff_base: Duration,
92 backoff_max: Duration,
93 defaults: Defaults,
94 validate: bool,
95 user_agent: Option<String>,
96}
97
98impl Default for HonkBuilder {
99 fn default() -> Self {
100 HonkBuilder {
101 url: String::new(),
102 key: String::new(),
103 timeout: DEFAULT_TIMEOUT,
104 retries: DEFAULT_RETRIES,
105 deadline: DEFAULT_DEADLINE,
106 backoff_base: DEFAULT_BACKOFF_BASE,
107 backoff_max: DEFAULT_BACKOFF_MAX,
108 defaults: Defaults::default(),
109 validate: true,
110 user_agent: None,
111 }
112 }
113}
114
115fn env_var(name: &str) -> Option<String> {
116 std::env::var(name).ok().filter(|v| !v.trim().is_empty())
117}
118
119impl HonkBuilder {
120 pub fn new() -> Self {
122 HonkBuilder::default()
123 }
124
125 pub fn from_env() -> Self {
128 HonkBuilder {
129 url: env_var("HONK_URL").unwrap_or_default(),
130 key: env_var("HONK_KEY").unwrap_or_default(),
131 defaults: Defaults {
132 source: env_var("HONK_SOURCE"),
133 environment: env_var("HONK_ENVIRONMENT"),
134 channel: env_var("HONK_CHANNEL"),
135 },
136 ..HonkBuilder::default()
137 }
138 }
139
140 pub fn url(mut self, url: impl Into<String>) -> Self {
142 self.url = url.into();
143 self
144 }
145
146 pub fn key(mut self, key: impl Into<String>) -> Self {
148 self.key = key.into();
149 self
150 }
151
152 pub fn timeout(mut self, timeout: Duration) -> Self {
154 self.timeout = timeout;
155 self
156 }
157
158 pub fn retries(mut self, retries: u32) -> Self {
161 self.retries = retries;
162 self
163 }
164
165 pub fn deadline(mut self, deadline: Duration) -> Self {
168 self.deadline = deadline;
169 self
170 }
171
172 pub fn backoff(mut self, base: Duration, max: Duration) -> Self {
176 self.backoff_base = base;
177 self.backoff_max = max;
178 self
179 }
180
181 pub fn defaults(mut self, defaults: Defaults) -> Self {
183 self.defaults = defaults;
184 self
185 }
186
187 pub fn source(mut self, source: impl Into<String>) -> Self {
189 self.defaults.source = Some(source.into());
190 self
191 }
192
193 pub fn environment(mut self, environment: impl Into<String>) -> Self {
195 self.defaults.environment = Some(environment.into());
196 self
197 }
198
199 pub fn channel(mut self, channel: impl Into<String>) -> Self {
201 self.defaults.channel = Some(channel.into());
202 self
203 }
204
205 pub fn validate(mut self, validate: bool) -> Self {
208 self.validate = validate;
209 self
210 }
211
212 pub fn user_agent(mut self, user_agent: impl Into<String>) -> Self {
214 self.user_agent = Some(user_agent.into());
215 self
216 }
217
218 pub fn build(self) -> Result<Honk> {
220 let config = |msg: &str| Error::InvalidConfiguration(msg.to_owned());
221 let mut url = self.url.trim().trim_end_matches('/').to_owned();
222 if let Some(base) = url.strip_suffix("/v1/messages") {
223 url = base.to_owned();
224 }
225 if url.is_empty() {
226 return Err(config(
227 "URL is required (the base address of your Honk server, e.g. https://honk-honk.app; is HONK_URL set?)",
228 ));
229 }
230 let lower = url.to_ascii_lowercase();
231 if !(lower.starts_with("https://") || lower.starts_with("http://"))
232 || reqwest::Url::parse(&url).is_err()
233 {
234 return Err(Error::InvalidConfiguration(format!(
235 "URL must be an https:// address (got {:?})",
236 self.url
237 )));
238 }
239 let key = self.key.trim().to_owned();
240 if key.is_empty() {
241 return Err(config(
242 "key is required (a project ingestion key honk_…; is HONK_KEY set?)",
243 ));
244 }
245 if !key.starts_with("honk_") || !key.bytes().all(|b| (0x21..=0x7e).contains(&b)) {
246 return Err(config(
247 "key must be a project ingestion key starting with honk_ (create one under Project → Keys)",
248 ));
249 }
250 let http = reqwest::Client::builder()
251 .redirect(reqwest::redirect::Policy::none())
252 .pool_idle_timeout(Duration::from_secs(60))
253 .build()
254 .map_err(|e| {
255 Error::InvalidConfiguration(format!("could not create the HTTP client: {e}"))
256 })?;
257 let mut user_agent = format!("honk-rust/{VERSION}");
258 if let Some(extra) = self.user_agent.filter(|s| !s.trim().is_empty()) {
259 user_agent.push(' ');
260 user_agent.push_str(extra.trim());
261 }
262 Ok(Honk {
263 inner: Arc::new(Inner {
264 endpoint: format!("{url}/v1/messages"),
265 key,
266 timeout: if self.timeout.is_zero() {
267 DEFAULT_TIMEOUT
268 } else {
269 self.timeout
270 },
271 retries: self.retries,
272 deadline: if self.deadline.is_zero() {
273 DEFAULT_DEADLINE
274 } else {
275 self.deadline
276 },
277 backoff_base: if self.backoff_base.is_zero() {
278 DEFAULT_BACKOFF_BASE
279 } else {
280 self.backoff_base
281 },
282 backoff_max: if self.backoff_max.is_zero() {
283 DEFAULT_BACKOFF_MAX
284 } else {
285 self.backoff_max
286 },
287 defaults: self.defaults,
288 validate: self.validate,
289 user_agent,
290 http,
291 }),
292 })
293 }
294
295 #[cfg(feature = "blocking")]
297 #[cfg_attr(docsrs, doc(cfg(feature = "blocking")))]
298 pub fn build_blocking(self) -> Result<crate::blocking::Honk> {
299 crate::blocking::Honk::from_async(self.build()?)
300 }
301}
302
303impl Honk {
304 pub fn new(url: impl Into<String>, key: impl Into<String>) -> Result<Honk> {
307 HonkBuilder::new().url(url).key(key).build()
308 }
309
310 pub fn from_env() -> Result<Honk> {
313 HonkBuilder::from_env().build()
314 }
315
316 pub fn builder() -> HonkBuilder {
318 HonkBuilder::new()
319 }
320
321 pub async fn send<'k>(
344 &self,
345 message: &Message,
346 idempotency_key: impl Into<Option<&'k str>>,
347 ) -> Result<Accepted> {
348 let me = &*self.inner;
349 let body = encode_message(message, &me.defaults, me.validate)?;
350 let key = match idempotency_key.into() {
351 None => crate::uuid::new_idempotency_key(),
352 Some(k) if valid_idempotency_key(k) => k.to_owned(),
353 Some(_) => {
354 return Err(Error::local_validation(vec![FieldError::new(
355 "Idempotency-Key",
356 "invalid_format",
357 "use 1-128 printable ASCII characters without spaces",
358 )]));
359 }
360 };
361
362 let deadline = Instant::now() + me.deadline;
363 let mut attempt: u32 = 1;
364 loop {
365 let remaining = deadline.saturating_duration_since(Instant::now());
366 let timeout = me.timeout.min(remaining).max(Duration::from_millis(1));
367 let failure = match self.attempt(&body, &key, timeout, attempt).await {
368 Ok(accepted) => return Ok(accepted),
369 Err(e) => e,
370 };
371 if !failure.is_retryable() || attempt > me.retries {
372 return Err(failure);
373 }
374 let mut wait = self.jitter(attempt);
375 if let Some(after) = failure.retry_after() {
376 wait = wait.max(after);
377 }
378 if Instant::now() + wait >= deadline {
379 return Err(failure);
380 }
381 tokio::time::sleep(wait).await;
382 attempt += 1;
383 }
384 }
385
386 pub async fn honk<'k>(
402 &self,
403 message: &Message,
404 idempotency_key: impl Into<Option<&'k str>>,
405 ) -> Result<Accepted> {
406 self.send(message, idempotency_key).await
407 }
408
409 fn jitter(&self, attempt: u32) -> Duration {
410 let me = &*self.inner;
411 let exp = me
412 .backoff_base
413 .saturating_mul(1u32.checked_shl(attempt - 1).unwrap_or(u32::MAX));
414 let ceiling = exp.min(me.backoff_max).as_micros() as u64;
415 Duration::from_micros(crate::uuid::random_u64() % (ceiling + 1))
416 }
417
418 async fn attempt(
419 &self,
420 body: &str,
421 key: &str,
422 timeout: Duration,
423 attempt: u32,
424 ) -> Result<Accepted> {
425 let me = &*self.inner;
426 let failure = |kind: fn(Box<Failure>) -> Error,
427 code: &str,
428 message: String,
429 source: reqwest::Error| {
430 kind(Box::new(Failure {
431 code: code.into(),
432 message,
433 idempotency_key: Some(key.to_owned()),
434 attempts: attempt,
435 source: Some(Box::new(source)),
436 ..Failure::default()
437 }))
438 };
439 let transport = |e: reqwest::Error| {
440 if e.is_timeout() {
441 failure(
442 Error::Timeout,
443 "timeout",
444 format!(
445 "no answer within {timeout:?} (the event may or may not have been stored; retrying with the same idempotency key is safe)"
446 ),
447 e,
448 )
449 } else {
450 let msg = format!("could not reach Honk: {}", error_chain(&e));
451 failure(Error::Network, "network_error", msg, e)
452 }
453 };
454
455 let mut res = me
456 .http
457 .post(&me.endpoint)
458 .header(AUTHORIZATION, format!("Bearer {}", me.key))
459 .header("Idempotency-Key", key)
460 .header(CONTENT_TYPE, "application/json")
461 .header(ACCEPT, "application/json")
462 .header(USER_AGENT, &me.user_agent)
463 .body(body.to_owned())
464 .timeout(timeout)
465 .send()
466 .await
467 .map_err(transport)?;
468 let status = res.status().as_u16();
469 let headers = res.headers().clone();
470 let mut data = Vec::new();
471 while let Some(chunk) = res.chunk().await.map_err(transport)? {
472 data.extend_from_slice(&chunk[..chunk.len().min(MAX_ANSWER_BYTES - data.len())]);
473 if data.len() >= MAX_ANSWER_BYTES {
474 break;
475 }
476 }
477 if (200..300).contains(&status) {
478 return match serde_json::from_slice::<Accepted>(&data) {
479 Ok(a) if !a.id.is_empty() => Ok(a),
480 _ => Err(Error::Http(Box::new(Failure {
481 status: Some(status),
482 message: "answer without a message id".into(),
483 idempotency_key: Some(key.to_owned()),
484 attempts: attempt,
485 ..Failure::default()
486 }))),
487 };
488 }
489 Err(response_error(status, &headers, &data, key, attempt))
490 }
491
492 pub fn problem(
503 &self,
504 group_key: impl Into<String>,
505 title: impl Into<String>,
506 message: impl Into<String>,
507 ) -> PendingSend<'_> {
508 self.pending(Severity::Error, title, message)
509 .group_key(group_key)
510 .event_type(EventType::Problem)
511 }
512
513 pub fn recovery(
516 &self,
517 group_key: impl Into<String>,
518 title: impl Into<String>,
519 message: impl Into<String>,
520 ) -> PendingSend<'_> {
521 self.pending(Severity::Success, title, message)
522 .group_key(group_key)
523 .event_type(EventType::Recovery)
524 }
525
526 pub fn uhoh(
538 &self,
539 group_key: impl Into<String>,
540 title: impl Into<String>,
541 message: impl Into<String>,
542 ) -> PendingSend<'_> {
543 self.problem(group_key, title, message)
544 }
545
546 pub fn phew(
549 &self,
550 group_key: impl Into<String>,
551 title: impl Into<String>,
552 message: impl Into<String>,
553 ) -> PendingSend<'_> {
554 self.recovery(group_key, title, message)
555 }
556
557 pub fn light(&self, title: impl Into<String>, message: impl Into<String>) -> PendingSend<'_> {
574 self.pending(Severity::Info, title, message)
575 }
576
577 pub fn beep(&self, title: impl Into<String>, message: impl Into<String>) -> PendingSend<'_> {
579 self.pending(Severity::Success, title, message)
580 }
581
582 pub fn loud(&self, title: impl Into<String>, message: impl Into<String>) -> PendingSend<'_> {
584 self.pending(Severity::Warning, title, message)
585 }
586
587 pub fn long(&self, title: impl Into<String>, message: impl Into<String>) -> PendingSend<'_> {
589 self.pending(Severity::Error, title, message)
590 }
591
592 pub fn blast(&self, title: impl Into<String>, message: impl Into<String>) -> PendingSend<'_> {
594 self.pending(Severity::Critical, title, message)
595 }
596
597 pub fn info(&self, title: impl Into<String>, message: impl Into<String>) -> PendingSend<'_> {
599 self.light(title, message)
600 }
601
602 pub fn success(&self, title: impl Into<String>, message: impl Into<String>) -> PendingSend<'_> {
604 self.beep(title, message)
605 }
606
607 pub fn warning(&self, title: impl Into<String>, message: impl Into<String>) -> PendingSend<'_> {
609 self.loud(title, message)
610 }
611
612 pub fn error(&self, title: impl Into<String>, message: impl Into<String>) -> PendingSend<'_> {
614 self.long(title, message)
615 }
616
617 pub fn critical(
619 &self,
620 title: impl Into<String>,
621 message: impl Into<String>,
622 ) -> PendingSend<'_> {
623 self.blast(title, message)
624 }
625
626 fn pending(
627 &self,
628 severity: Severity,
629 title: impl Into<String>,
630 message: impl Into<String>,
631 ) -> PendingSend<'_> {
632 let title: String = title.into();
633 let mut message = Message::new(message).severity(severity);
634 if !title.is_empty() {
635 message.title = Some(title);
636 }
637 PendingSend {
638 honk: self,
639 message,
640 idempotency_key: None,
641 }
642 }
643}
644
645fn error_chain(e: &dyn std::error::Error) -> String {
647 let mut out = e.to_string();
648 let mut next = e.source();
649 while let Some(s) = next {
650 let text = s.to_string();
651 if !out.contains(&text) {
652 out.push_str(": ");
653 out.push_str(&text);
654 }
655 next = s.source();
656 }
657 out
658}
659
660pub(crate) fn parse_retry_after(value: &str, now: SystemTime) -> Option<Duration> {
662 let v = value.trim();
663 if v.is_empty() {
664 return None;
665 }
666 if let Ok(secs) = v.parse::<u64>() {
667 return Some(Duration::from_secs(secs));
668 }
669 let at = httpdate::parse_http_date(v).ok()?;
670 at.duration_since(now)
671 .ok()
672 .map(|d| Duration::from_secs(d.as_secs() + u64::from(d.subsec_millis() >= 500)))
673}
674
675fn response_error(
676 status: u16,
677 headers: &reqwest::header::HeaderMap,
678 data: &[u8],
679 key: &str,
680 attempt: u32,
681) -> Error {
682 #[derive(serde::Deserialize, Default)]
683 struct Envelope {
684 #[serde(default)]
685 error: Body,
686 }
687 #[derive(serde::Deserialize, Default)]
688 struct Body {
689 #[serde(default)]
690 code: String,
691 #[serde(default)]
692 message: String,
693 #[serde(default)]
694 request_id: String,
695 #[serde(default)]
696 fields: Vec<FieldError>,
697 }
698 let header = |name| {
699 headers
700 .get(name)
701 .and_then(|v: &reqwest::header::HeaderValue| v.to_str().ok())
702 };
703 let env: Envelope = serde_json::from_slice(data).unwrap_or_default();
704 let mut message = env.error.message;
705 if message.is_empty() {
706 message = String::from_utf8_lossy(data)
707 .trim()
708 .chars()
709 .take(200)
710 .collect();
711 }
712 if message.is_empty() {
713 message = reqwest::StatusCode::from_u16(status)
714 .ok()
715 .and_then(|s| s.canonical_reason())
716 .unwrap_or("unexpected answer")
717 .to_owned();
718 }
719 let request_id = Some(env.error.request_id)
720 .filter(|s| !s.is_empty())
721 .or_else(|| header("x-request-id").map(str::to_owned));
722 let kind: fn(Box<Failure>) -> Error = match status {
723 400 | 413 | 415 | 422 => Error::Validation,
724 401 | 403 => Error::Auth,
725 409 => Error::Conflict,
726 429 => Error::Quota,
727 500.. => Error::Server,
728 300..=399 => {
729 message = match header(LOCATION.as_str()) {
730 Some(loc) => format!("redirect to {loc}; set the URL to the final https address"),
731 None => "redirect; set the URL to the final https address".into(),
732 };
733 Error::Http
734 }
735 404 => {
736 message.push_str(" (is the URL the base address of your Honk server?)");
737 Error::Http
738 }
739 _ => Error::Http,
740 };
741 kind(Box::new(Failure {
742 status: Some(status),
743 code: env.error.code,
744 message,
745 fields: env.error.fields,
746 request_id,
747 idempotency_key: Some(key.to_owned()),
748 attempts: attempt,
749 retry_after: header(RETRY_AFTER.as_str())
750 .and_then(|v| parse_retry_after(v, SystemTime::now())),
751 ..Failure::default()
752 }))
753}
754
755#[must_use = "a PendingSend does nothing until it is awaited"]
758pub struct PendingSend<'a> {
759 honk: &'a Honk,
760 message: Message,
761 idempotency_key: Option<String>,
762}
763
764impl std::fmt::Debug for PendingSend<'_> {
765 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
766 f.debug_struct("PendingSend")
767 .field("message", &self.message)
768 .field("idempotency_key", &self.idempotency_key)
769 .finish()
770 }
771}
772
773macro_rules! pending_setters {
774 ($($(#[$doc:meta])* $name:ident: $t:ty $(=> $conv:ident)?;)+) => {
775 $(
776 $(#[$doc])*
777 pub fn $name(mut self, value: $t) -> Self {
778 self.message = self.message.$name(value $(.$conv())?);
779 self
780 }
781 )+
782 };
783}
784
785impl<'a> PendingSend<'a> {
786 pub fn idempotency_key(mut self, key: impl Into<String>) -> Self {
788 self.idempotency_key = Some(key.into());
789 self
790 }
791
792 pending_setters! {
793 severity: Severity;
795 priority: Priority;
797 category: Category;
799 source: impl Into<String> => into;
801 environment: impl Into<String> => into;
803 channel: impl Into<String> => into;
805 group_key: impl Into<String> => into;
807 event_type: EventType;
809 occurred_at: SystemTime;
811 url: impl Into<String> => into;
813 image_url: impl Into<String> => into;
815 ttl_seconds: u32;
817 source_sequence: u64;
819 }
820
821 pub fn action(mut self, title: impl Into<String>, url: impl Into<String>) -> Self {
823 self.message = self.message.action(title, url);
824 self
825 }
826
827 pub fn metadata(mut self, key: impl Into<String>, value: impl Into<MetadataValue>) -> Self {
829 self.message = self.message.metadata(key, value);
830 self
831 }
832
833 pub fn message(&self) -> &Message {
835 &self.message
836 }
837
838 pub async fn send(self) -> Result<Accepted> {
840 self.honk
841 .send(&self.message, self.idempotency_key.as_deref())
842 .await
843 }
844
845 #[cfg(feature = "blocking")]
846 pub(crate) fn into_parts(self) -> (Message, Option<String>) {
847 (self.message, self.idempotency_key)
848 }
849}
850
851impl<'a> IntoFuture for PendingSend<'a> {
852 type Output = Result<Accepted>;
853 type IntoFuture = Pin<Box<dyn Future<Output = Result<Accepted>> + Send + 'a>>;
854
855 fn into_future(self) -> Self::IntoFuture {
856 Box::pin(self.send())
857 }
858}
859
860#[cfg(test)]
861mod tests {
862 use super::*;
863 use std::time::UNIX_EPOCH;
864
865 #[test]
866 fn retry_after_seconds_and_http_dates() {
867 let now = UNIX_EPOCH + Duration::from_secs(1_791_116_405);
868 assert_eq!(parse_retry_after("7", now), Some(Duration::from_secs(7)));
869 assert_eq!(parse_retry_after(" 0 ", now), Some(Duration::ZERO));
870 assert_eq!(
871 parse_retry_after(&httpdate::fmt_http_date(now + Duration::from_secs(90)), now),
872 Some(Duration::from_secs(90))
873 );
874 assert_eq!(
875 parse_retry_after(&httpdate::fmt_http_date(now - Duration::from_secs(90)), now),
876 None
877 );
878 assert_eq!(parse_retry_after("soon", now), None);
879 assert_eq!(parse_retry_after("-3", now), None);
880 assert_eq!(parse_retry_after("", now), None);
881 }
882
883 #[test]
884 fn jitter_stays_under_the_ceiling() {
885 let honk = Honk::builder()
886 .url("https://honk.example.com")
887 .key("honk_ab12cd34ef56_ABCDEFGHIJKLMNOPQRSTUVWXYZ012345")
888 .backoff(Duration::from_millis(100), Duration::from_millis(250))
889 .build()
890 .unwrap();
891 for attempt in 1..40 {
892 let ceiling = Duration::from_millis(100 * (1u64 << (attempt - 1).min(20)))
893 .min(Duration::from_millis(250));
894 assert!(honk.jitter(attempt) <= ceiling, "attempt {attempt}");
895 }
896 }
897}