1#![doc = include_str!("../README.md")]
17#![forbid(unsafe_code)]
18
19mod clock;
20mod deliver;
21mod store;
22
23pub use crate::deliver::{
24 DeliveryError, DrainReport, EVENT_ID_HEADER, EVENT_TYPE_HEADER, SIGNATURE_HEADER,
25};
26pub use crate::store::{DeadLetter, Delivery, Endpoint, TOPIC_DELIVER};
27use cratefield_core::{
28 AnyError, BoxFuture, Config, ConfigError, Database, DbError, IdGen, Migrations, Module,
29 ModuleConfig, ModuleContext, PersonalDataSet, Port, SqlMigration, UlidIdGen,
30};
31
32pub const MODULE_NAME: &str = "webhooks";
34
35const MIGRATION_INIT: SqlMigration = SqlMigration::new(
38 "0001",
39 "init",
40 include_str!("../migrations/sqlite/0001_init.sql"),
41);
42
43const MAX_ATTEMPTS_RANGE: std::ops::RangeInclusive<u32> = 1..=1_000;
46
47#[derive(Debug, Clone)]
49pub struct Webhooks {
50 max_attempts: u32,
53}
54
55impl Default for Webhooks {
56 fn default() -> Self {
57 Self::new()
58 }
59}
60
61impl Webhooks {
62 #[must_use]
64 pub fn new() -> Self {
65 Self { max_attempts: 5 }
66 }
67
68 #[must_use]
73 pub fn max_attempts(mut self, max_attempts: u32) -> Self {
74 self.max_attempts =
75 max_attempts.clamp(*MAX_ATTEMPTS_RANGE.start(), *MAX_ATTEMPTS_RANGE.end());
76 self
77 }
78
79 pub async fn create_endpoint(
94 &self,
95 db: &dyn Database,
96 subject: &str,
97 url: &str,
98 event_types: &[&str],
99 now: &str,
100 ) -> Result<CreatedEndpoint, EndpointError> {
101 if subject.trim().is_empty() || subject.len() > 255 {
102 return Err(EndpointError::InvalidSubject);
103 }
104 if !is_http_url(url) {
105 return Err(EndpointError::InvalidUrl(url.to_owned()));
106 }
107 for candidate in event_types {
108 let candidate = candidate.trim();
109 if !candidate.is_empty() && !is_valid_event_type(candidate) {
110 return Err(EndpointError::InvalidEventType(candidate.to_owned()));
111 }
112 }
113 let secret = generate_secret()?;
114 let id = UlidIdGen.ulid();
115 db.execute(&store::insert_endpoint_statement(
116 &id,
117 subject,
118 url,
119 &secret,
120 &normalize_event_types(event_types),
121 now,
122 ))
123 .await?;
124 Ok(CreatedEndpoint {
125 endpoint_id: id,
126 secret,
127 })
128 }
129
130 pub async fn endpoints(
137 &self,
138 db: &dyn Database,
139 subject: &str,
140 ) -> Result<Vec<Endpoint>, DbError> {
141 store::endpoints_for_subject(db, subject).await
142 }
143
144 pub async fn delete_endpoint(
152 &self,
153 db: &dyn Database,
154 subject: &str,
155 endpoint_id: &str,
156 ) -> Result<bool, DbError> {
157 store::delete_endpoint(db, subject, endpoint_id).await
158 }
159
160 pub async fn publish(
178 &self,
179 db: &dyn Database,
180 subject: &str,
181 event_type: &str,
182 data: &serde_json::Value,
183 now: &str,
184 ) -> Result<Published, PublishError> {
185 if !is_valid_event_type(event_type) {
186 return Err(PublishError::InvalidEventType(event_type.to_owned()));
187 }
188 let endpoints = store::endpoints_for_subject(db, subject).await?;
189 let event_id = UlidIdGen.ulid();
190 let mut statements = Vec::new();
191 for endpoint in &endpoints {
192 if !matches_filter(&endpoint.event_types, event_type) {
193 continue;
194 }
195 statements.push(store::enqueue_statement(
196 &store::DeliveryJob {
197 endpoint_id: endpoint.id.clone(),
198 event_id: event_id.clone(),
199 event_type: event_type.to_owned(),
200 subject: subject.to_owned(),
201 created_at: now.to_owned(),
202 data: data.clone(),
203 },
204 subject,
205 now,
206 ));
207 }
208 Ok(Published {
209 event_id,
210 endpoints: statements.len(),
211 statements,
212 })
213 }
214
215 pub async fn drain_with(&self, ctx: &ModuleContext) -> Result<DrainReport, DeliveryError> {
234 deliver::drain(ctx, self.max_attempts).await
235 }
236
237 pub async fn replay(
248 &self,
249 db: &dyn Database,
250 subject: &str,
251 dead_letter_id: &str,
252 now: &str,
253 ) -> Result<bool, DbError> {
254 store::replay(db, subject, dead_letter_id, now).await
255 }
256
257 pub async fn deliveries(
264 &self,
265 db: &dyn Database,
266 subject: &str,
267 limit: u64,
268 ) -> Result<Vec<Delivery>, DbError> {
269 store::deliveries(db, subject, limit).await
270 }
271
272 pub async fn dead_letters(
279 &self,
280 db: &dyn Database,
281 subject: &str,
282 limit: u64,
283 ) -> Result<Vec<DeadLetter>, DbError> {
284 store::dead_letters(db, subject, limit).await
285 }
286}
287
288impl Module for Webhooks {
289 fn name(&self) -> &'static str {
290 MODULE_NAME
291 }
292
293 fn version(&self) -> &'static str {
294 env!("CARGO_PKG_VERSION")
295 }
296
297 fn requires(&self) -> &'static [Port] {
298 &[Port::Db, Port::HttpClient, Port::Clock]
301 }
302
303 fn tables(&self) -> &'static [&'static str] {
304 &[
305 store::ENDPOINTS,
306 store::OUTBOX,
307 store::DELIVERIES,
308 store::DEAD_LETTERS,
309 ]
310 }
311
312 fn personal_data(&self) -> &'static [PersonalDataSet] {
316 const SETS: &[PersonalDataSet] = &[
317 PersonalDataSet {
318 table: store::ENDPOINTS,
319 subject: "subject",
320 kind: cratefield_core::DataKind::Identifier,
321 disposition: cratefield_core::Disposition::Erase,
322 description: "The webhook URLs this venture delivers your events to, one row \
323 each. Each row holds a signing secret, which is why it is \
324 redacted here; deleting your account deletes the endpoints and \
325 their secrets.",
326 redacted: &["secret"],
327 subject_via: None,
328 },
329 PersonalDataSet {
330 table: store::OUTBOX,
331 subject: "subject",
332 kind: cratefield_core::DataKind::Content,
333 disposition: cratefield_core::Disposition::Erase,
334 description: "Events queued for delivery to the endpoints above, with the \
335 account they are about. A delivery is erased with the account \
336 even if it has not been sent yet.",
337 redacted: &[],
338 subject_via: None,
339 },
340 PersonalDataSet {
341 table: store::DELIVERIES,
342 subject: "subject",
343 kind: cratefield_core::DataKind::Usage,
344 disposition: cratefield_core::Disposition::Erase,
345 description: "The log of delivery attempts: which endpoint, which event, what \
346 status code came back, and when. It shows which of your events \
347 left the venture, and nothing else.",
348 redacted: &[],
349 subject_via: None,
350 },
351 PersonalDataSet {
352 table: store::DEAD_LETTERS,
353 subject: "subject",
354 kind: cratefield_core::DataKind::Content,
355 disposition: cratefield_core::Disposition::Erase,
356 description: "Events that gave up after repeated failed deliveries, kept so \
357 they can be replayed. They carry the same data as the queued \
358 events above, and go when your account is deleted.",
359 redacted: &[],
360 subject_via: None,
361 },
362 ];
363 SETS
364 }
365
366 fn migrations(&self) -> Migrations {
367 const MIGRATIONS: [SqlMigration; 1] = [MIGRATION_INIT];
368 const _: () = cratefield_core::assert_migration_set(&MIGRATIONS);
371 Migrations {
372 sqlite: &MIGRATIONS,
373 postgres: &[],
374 }
375 }
376
377 fn validate_config(&self, cfg: &dyn Config) -> Result<(), ConfigError> {
378 let module = ModuleConfig::new(MODULE_NAME, cfg);
379 let mut errors = ConfigError::default();
380 if let Some(raw) = cfg
381 .get(&module.key("MAX_ATTEMPTS"))
382 .map(|raw| raw.parse::<u32>())
383 && !raw.is_ok_and(|value| MAX_ATTEMPTS_RANGE.contains(&value))
384 {
385 errors.push(format!(
386 "webhooks: {} must be a whole number between 1 and 1000",
387 module.key("MAX_ATTEMPTS")
388 ));
389 }
390 errors.into_result()
391 }
392
393 fn router(&self, _ctx: ModuleContext) -> axum::Router {
397 axum::Router::new()
398 }
399
400 fn scheduled<'a>(
406 &'a self,
407 ctx: &'a ModuleContext,
408 cron: &'a str,
409 ) -> BoxFuture<'a, Result<(), AnyError>> {
410 Box::pin(async move {
411 let report = self
412 .drain_with(ctx)
413 .await
414 .map_err(|err| Box::new(err) as AnyError)?;
415 if report.claimed > 0 || report.deferred > 0 {
418 tracing::info!(?report, cron, "drained the webhooks outbox");
419 }
420 Ok(())
421 })
422 }
423}
424
425#[derive(Clone, PartialEq, Eq)]
429pub struct CreatedEndpoint {
430 pub endpoint_id: String,
431 pub secret: String,
432}
433
434impl std::fmt::Debug for CreatedEndpoint {
438 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
439 f.debug_struct("CreatedEndpoint")
440 .field("endpoint_id", &self.endpoint_id)
441 .field("secret", &"[redacted]")
442 .finish()
443 }
444}
445
446#[derive(Debug, Clone, PartialEq, Eq)]
448pub enum EndpointError {
449 InvalidUrl(String),
452 InvalidSubject,
454 InvalidEventType(String),
458 Entropy(String),
461 Database(DbError),
463}
464
465impl std::fmt::Display for EndpointError {
466 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
467 match self {
468 EndpointError::InvalidUrl(url) => {
469 write!(f, "webhooks: not an http(s) URL: {url:?}")
470 }
471 EndpointError::InvalidSubject => {
472 f.write_str("webhooks: the subject must be 1..=255 bytes")
473 }
474 EndpointError::InvalidEventType(event_type) => {
475 write!(f, "webhooks: not a usable event type: {event_type:?}")
476 }
477 EndpointError::Entropy(error) => write!(f, "webhooks: {error}"),
478 EndpointError::Database(error) => write!(f, "webhooks: {error}"),
479 }
480 }
481}
482
483impl std::error::Error for EndpointError {}
484
485impl From<DbError> for EndpointError {
486 fn from(error: DbError) -> Self {
487 EndpointError::Database(error)
488 }
489}
490
491#[derive(Debug, Clone, PartialEq, Eq)]
493pub enum PublishError {
494 InvalidEventType(String),
499 Database(DbError),
501}
502
503impl std::fmt::Display for PublishError {
504 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
505 match self {
506 PublishError::InvalidEventType(event_type) => {
507 write!(f, "webhooks: not a usable event type: {event_type:?}")
508 }
509 PublishError::Database(error) => write!(f, "webhooks: {error}"),
510 }
511 }
512}
513
514impl std::error::Error for PublishError {}
515
516impl From<DbError> for PublishError {
517 fn from(error: DbError) -> Self {
518 PublishError::Database(error)
519 }
520}
521
522#[derive(Debug, Clone)]
526pub struct Published {
527 pub event_id: String,
531 pub endpoints: usize,
534 statements: Vec<cratefield_core::Statement>,
535}
536
537impl Published {
538 #[must_use]
541 pub fn into_statements(self) -> Vec<cratefield_core::Statement> {
542 self.statements
543 }
544}
545
546fn is_valid_event_type(event_type: &str) -> bool {
552 (1..=128).contains(&event_type.len()) && event_type.chars().all(|c| c.is_ascii_graphic())
553}
554
555#[must_use]
561pub fn matches_filter(filter: &str, event_type: &str) -> bool {
562 let filter = filter.trim();
563 filter == "*"
564 || filter
565 .split(',')
566 .map(str::trim)
567 .any(|candidate| candidate == event_type)
568}
569
570#[must_use]
573pub fn normalize_event_types(event_types: &[&str]) -> String {
574 let mut seen = Vec::new();
575 for candidate in event_types {
576 let candidate = candidate.trim();
577 if !candidate.is_empty() && !seen.contains(&candidate) {
578 seen.push(candidate);
579 }
580 }
581 if seen.is_empty() {
582 "*".to_owned()
583 } else {
584 seen.join(",")
585 }
586}
587
588fn is_http_url(url: &str) -> bool {
593 let max = 2048;
594 (url.starts_with("https://") || url.starts_with("http://"))
595 && url.len() <= max
596 && !url.contains(char::is_whitespace)
597 && match url.split_once("://") {
598 Some((_, rest)) => {
601 let authority = rest.split(['/', '?', '#']).next().unwrap_or("");
602 !authority.is_empty()
603 }
604 None => false,
605 }
606}
607
608fn generate_secret() -> Result<String, EndpointError> {
613 let mut bytes = [0_u8; 32];
614 getrandom::fill(&mut bytes).map_err(|error| EndpointError::Entropy(error.to_string()))?;
615 let mut hex = String::with_capacity(2 * bytes.len());
616 for byte in bytes {
617 use std::fmt::Write as _;
619 let _ = write!(hex, "{byte:02x}");
620 }
621 Ok(format!("whsec_{hex}"))
622}
623
624#[cfg(test)]
625mod tests {
626 use super::*;
627
628 #[test]
629 fn defaults_are_the_documented_ones() {
630 let module = Webhooks::new();
631 assert_eq!(module.name(), "webhooks");
632 assert_eq!(module.version(), env!("CARGO_PKG_VERSION"));
633 assert_eq!(module.requires(), [Port::Db, Port::HttpClient, Port::Clock]);
634 assert_eq!(
635 module.tables(),
636 [
637 "webhooks_endpoints",
638 "webhooks_outbox",
639 "webhooks_deliveries",
640 "webhooks_dead_letters"
641 ]
642 );
643 assert_eq!(module.max_attempts, 5);
644 }
645
646 #[test]
647 fn max_attempts_stays_within_the_range_the_config_must_pass() {
648 assert_eq!(Webhooks::new().max_attempts(0).max_attempts, 1);
649 assert_eq!(Webhooks::new().max_attempts(2).max_attempts, 2);
650 assert_eq!(Webhooks::new().max_attempts(1_000).max_attempts, 1_000);
651 assert_eq!(Webhooks::new().max_attempts(5_000).max_attempts, 1_000);
652 }
653
654 #[test]
655 fn a_created_endpoints_debug_never_carries_the_secret() {
656 let created = CreatedEndpoint {
657 endpoint_id: "ep_01HXYZ".to_owned(),
658 secret: "whsec_hunter2".to_owned(),
659 };
660 let rendered = format!("{created:?}");
661 assert!(rendered.contains("ep_01HXYZ"), "{rendered}");
662 assert!(rendered.contains("[redacted]"), "{rendered}");
663 assert!(!rendered.contains("hunter2"), "{rendered}");
664 }
665
666 #[test]
667 fn event_types_must_be_header_safe_tokens() {
668 assert!(is_valid_event_type("order.paid"));
669 assert!(is_valid_event_type("order-paid_v2"));
670 assert!(!is_valid_event_type(""), "empty");
671 assert!(!is_valid_event_type(" order.paid"), "leading space");
672 assert!(!is_valid_event_type("order paid"), "inner space");
673 assert!(!is_valid_event_type("command\ninjection"), "control char");
674 assert!(!is_valid_event_type("héllo"), "non-ASCII");
675 assert!(is_valid_event_type(&"x".repeat(128)), "the max is allowed");
676 assert!(!is_valid_event_type(&"x".repeat(129)), "over the max");
677 }
678
679 #[test]
680 fn a_star_filter_matches_everything_and_a_list_matches_exactly() {
681 assert!(matches_filter("*", "order.paid"));
682 assert!(matches_filter("*", ""));
683 assert!(matches_filter("order.paid", "order.paid"));
684 assert!(matches_filter(
685 " order.paid , order.refunded ",
686 "order.refunded"
687 ));
688 assert!(!matches_filter("order.paid", "order.paid.v2"));
689 assert!(!matches_filter("order.paid", "order.refunded"));
690 assert!(
691 !matches_filter("", "order.paid"),
692 "empty stores as *, never as nothing"
693 );
694 }
695
696 #[test]
697 fn filters_normalise_to_a_stored_list_or_a_star() {
698 assert_eq!(
699 normalize_event_types(&["order.paid", "order.refunded"]),
700 "order.paid,order.refunded"
701 );
702 assert_eq!(normalize_event_types(&[" order.paid ", ""]), "order.paid");
703 assert_eq!(normalize_event_types(&["a", "a", "b"]), "a,b");
704 assert_eq!(normalize_event_types(&[]), "*");
705 assert_eq!(normalize_event_types(&["", " "]), "*");
706 }
707
708 #[test]
709 fn urls_must_be_http_s_with_nowhere_for_whitespace_to_hide() {
710 assert!(is_http_url("https://example.com/hook"));
711 assert!(is_http_url("http://example.com"));
712 assert!(!is_http_url("example.com/hook"));
713 assert!(!is_http_url("ftp://example.com"));
714 assert!(!is_http_url("https://"));
715 assert!(!is_http_url("http:///path"), "no host at all");
716 assert!(!is_http_url("https:///q?a=1"), "no host, only a query");
717 assert!(is_http_url("http://example.com?a=1"), "authority, no path");
718 assert!(!is_http_url("https://example.com/a b"));
719 assert!(!is_http_url(""));
720 assert!(!is_http_url("https://example.com/\n"));
721 }
722
723 #[test]
724 fn secrets_are_prefixed_hex_and_never_repeat() {
725 let a = generate_secret().expect("os randomness");
726 let b = generate_secret().expect("os randomness");
727 assert!(a.starts_with("whsec_"), "{a}");
728 let hex = a.strip_prefix("whsec_").expect("prefix");
729 assert_eq!(hex.len(), 64, "32 bytes in hex");
730 assert!(
731 hex.chars()
732 .all(|c| c.is_ascii_hexdigit() && !c.is_ascii_uppercase())
733 );
734 assert_ne!(a, b, "two secrets are two draws");
735 }
736
737 #[test]
738 fn a_bad_config_is_collected_not_thrown_one_at_a_time() {
739 let ok = MapConfig::from_pairs([("WEBHOOKS_MAX_ATTEMPTS", "3")]);
740 assert!(Webhooks::new().validate_config(&ok).is_ok());
741
742 let empty = MapConfig::from_pairs([("WEBHOOKS_MAX_ATTEMPTS", "")]);
743 let error = Webhooks::new()
744 .validate_config(&empty)
745 .expect_err("an unparsable attempt count is refused");
746 assert!(
747 error.to_string().contains("WEBHOOKS_MAX_ATTEMPTS"),
748 "{error}"
749 );
750
751 let zero = MapConfig::from_pairs([("WEBHOOKS_MAX_ATTEMPTS", "0")]);
752 assert!(Webhooks::new().validate_config(&zero).is_err());
753
754 let ceiling = MapConfig::from_pairs([("WEBHOOKS_MAX_ATTEMPTS", "1000")]);
756 assert!(Webhooks::new().validate_config(&ceiling).is_ok());
757 let past = MapConfig::from_pairs([("WEBHOOKS_MAX_ATTEMPTS", "1001")]);
758 assert!(Webhooks::new().validate_config(&past).is_err());
759 }
760
761 use cratefield_core::MapConfig;
762}