#![doc = include_str!("../README.md")]
#![forbid(unsafe_code)]
mod clock;
mod deliver;
mod store;
pub use crate::deliver::{
DeliveryError, DrainReport, EVENT_ID_HEADER, EVENT_TYPE_HEADER, SIGNATURE_HEADER,
};
pub use crate::store::{DeadLetter, Delivery, Endpoint, TOPIC_DELIVER};
use cratefield_core::{
AnyError, BoxFuture, Config, ConfigError, Database, DbError, IdGen, Migrations, Module,
ModuleConfig, ModuleContext, PersonalDataSet, Port, SqlMigration, UlidIdGen,
};
pub const MODULE_NAME: &str = "webhooks";
const MIGRATION_INIT: SqlMigration = SqlMigration::new(
"0001",
"init",
include_str!("../migrations/sqlite/0001_init.sql"),
);
const MAX_ATTEMPTS_RANGE: std::ops::RangeInclusive<u32> = 1..=1_000;
#[derive(Debug, Clone)]
pub struct Webhooks {
max_attempts: u32,
}
impl Default for Webhooks {
fn default() -> Self {
Self::new()
}
}
impl Webhooks {
#[must_use]
pub fn new() -> Self {
Self { max_attempts: 5 }
}
#[must_use]
pub fn max_attempts(mut self, max_attempts: u32) -> Self {
self.max_attempts =
max_attempts.clamp(*MAX_ATTEMPTS_RANGE.start(), *MAX_ATTEMPTS_RANGE.end());
self
}
pub async fn create_endpoint(
&self,
db: &dyn Database,
subject: &str,
url: &str,
event_types: &[&str],
now: &str,
) -> Result<CreatedEndpoint, EndpointError> {
if subject.trim().is_empty() || subject.len() > 255 {
return Err(EndpointError::InvalidSubject);
}
if !is_http_url(url) {
return Err(EndpointError::InvalidUrl(url.to_owned()));
}
for candidate in event_types {
let candidate = candidate.trim();
if !candidate.is_empty() && !is_valid_event_type(candidate) {
return Err(EndpointError::InvalidEventType(candidate.to_owned()));
}
}
let secret = generate_secret()?;
let id = UlidIdGen.ulid();
db.execute(&store::insert_endpoint_statement(
&id,
subject,
url,
&secret,
&normalize_event_types(event_types),
now,
))
.await?;
Ok(CreatedEndpoint {
endpoint_id: id,
secret,
})
}
pub async fn endpoints(
&self,
db: &dyn Database,
subject: &str,
) -> Result<Vec<Endpoint>, DbError> {
store::endpoints_for_subject(db, subject).await
}
pub async fn delete_endpoint(
&self,
db: &dyn Database,
subject: &str,
endpoint_id: &str,
) -> Result<bool, DbError> {
store::delete_endpoint(db, subject, endpoint_id).await
}
pub async fn publish(
&self,
db: &dyn Database,
subject: &str,
event_type: &str,
data: &serde_json::Value,
now: &str,
) -> Result<Published, PublishError> {
if !is_valid_event_type(event_type) {
return Err(PublishError::InvalidEventType(event_type.to_owned()));
}
let endpoints = store::endpoints_for_subject(db, subject).await?;
let event_id = UlidIdGen.ulid();
let mut statements = Vec::new();
for endpoint in &endpoints {
if !matches_filter(&endpoint.event_types, event_type) {
continue;
}
statements.push(store::enqueue_statement(
&store::DeliveryJob {
endpoint_id: endpoint.id.clone(),
event_id: event_id.clone(),
event_type: event_type.to_owned(),
subject: subject.to_owned(),
created_at: now.to_owned(),
data: data.clone(),
},
subject,
now,
));
}
Ok(Published {
event_id,
endpoints: statements.len(),
statements,
})
}
pub async fn drain_with(&self, ctx: &ModuleContext) -> Result<DrainReport, DeliveryError> {
deliver::drain(ctx, self.max_attempts).await
}
pub async fn replay(
&self,
db: &dyn Database,
subject: &str,
dead_letter_id: &str,
now: &str,
) -> Result<bool, DbError> {
store::replay(db, subject, dead_letter_id, now).await
}
pub async fn deliveries(
&self,
db: &dyn Database,
subject: &str,
limit: u64,
) -> Result<Vec<Delivery>, DbError> {
store::deliveries(db, subject, limit).await
}
pub async fn dead_letters(
&self,
db: &dyn Database,
subject: &str,
limit: u64,
) -> Result<Vec<DeadLetter>, DbError> {
store::dead_letters(db, subject, limit).await
}
}
impl Module for Webhooks {
fn name(&self) -> &'static str {
MODULE_NAME
}
fn version(&self) -> &'static str {
env!("CARGO_PKG_VERSION")
}
fn requires(&self) -> &'static [Port] {
&[Port::Db, Port::HttpClient, Port::Clock]
}
fn tables(&self) -> &'static [&'static str] {
&[
store::ENDPOINTS,
store::OUTBOX,
store::DELIVERIES,
store::DEAD_LETTERS,
]
}
fn personal_data(&self) -> &'static [PersonalDataSet] {
const SETS: &[PersonalDataSet] = &[
PersonalDataSet {
table: store::ENDPOINTS,
subject: "subject",
kind: cratefield_core::DataKind::Identifier,
disposition: cratefield_core::Disposition::Erase,
description: "The webhook URLs this venture delivers your events to, one row \
each. Each row holds a signing secret, which is why it is \
redacted here; deleting your account deletes the endpoints and \
their secrets.",
redacted: &["secret"],
subject_via: None,
},
PersonalDataSet {
table: store::OUTBOX,
subject: "subject",
kind: cratefield_core::DataKind::Content,
disposition: cratefield_core::Disposition::Erase,
description: "Events queued for delivery to the endpoints above, with the \
account they are about. A delivery is erased with the account \
even if it has not been sent yet.",
redacted: &[],
subject_via: None,
},
PersonalDataSet {
table: store::DELIVERIES,
subject: "subject",
kind: cratefield_core::DataKind::Usage,
disposition: cratefield_core::Disposition::Erase,
description: "The log of delivery attempts: which endpoint, which event, what \
status code came back, and when. It shows which of your events \
left the venture, and nothing else.",
redacted: &[],
subject_via: None,
},
PersonalDataSet {
table: store::DEAD_LETTERS,
subject: "subject",
kind: cratefield_core::DataKind::Content,
disposition: cratefield_core::Disposition::Erase,
description: "Events that gave up after repeated failed deliveries, kept so \
they can be replayed. They carry the same data as the queued \
events above, and go when your account is deleted.",
redacted: &[],
subject_via: None,
},
];
SETS
}
fn migrations(&self) -> Migrations {
const MIGRATIONS: [SqlMigration; 1] = [MIGRATION_INIT];
const _: () = cratefield_core::assert_migration_set(&MIGRATIONS);
Migrations {
sqlite: &MIGRATIONS,
postgres: &[],
}
}
fn validate_config(&self, cfg: &dyn Config) -> Result<(), ConfigError> {
let module = ModuleConfig::new(MODULE_NAME, cfg);
let mut errors = ConfigError::default();
if let Some(raw) = cfg
.get(&module.key("MAX_ATTEMPTS"))
.map(|raw| raw.parse::<u32>())
&& !raw.is_ok_and(|value| MAX_ATTEMPTS_RANGE.contains(&value))
{
errors.push(format!(
"webhooks: {} must be a whole number between 1 and 1000",
module.key("MAX_ATTEMPTS")
));
}
errors.into_result()
}
fn router(&self, _ctx: ModuleContext) -> axum::Router {
axum::Router::new()
}
fn scheduled<'a>(
&'a self,
ctx: &'a ModuleContext,
cron: &'a str,
) -> BoxFuture<'a, Result<(), AnyError>> {
Box::pin(async move {
let report = self
.drain_with(ctx)
.await
.map_err(|err| Box::new(err) as AnyError)?;
if report.claimed > 0 {
tracing::info!(?report, cron, "drained the webhooks outbox");
}
Ok(())
})
}
}
#[derive(Clone, PartialEq, Eq)]
pub struct CreatedEndpoint {
pub endpoint_id: String,
pub secret: String,
}
impl std::fmt::Debug for CreatedEndpoint {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("CreatedEndpoint")
.field("endpoint_id", &self.endpoint_id)
.field("secret", &"[redacted]")
.finish()
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum EndpointError {
InvalidUrl(String),
InvalidSubject,
InvalidEventType(String),
Entropy(String),
Database(DbError),
}
impl std::fmt::Display for EndpointError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
EndpointError::InvalidUrl(url) => {
write!(f, "webhooks: not an http(s) URL: {url:?}")
}
EndpointError::InvalidSubject => {
f.write_str("webhooks: the subject must be 1..=255 bytes")
}
EndpointError::InvalidEventType(event_type) => {
write!(f, "webhooks: not a usable event type: {event_type:?}")
}
EndpointError::Entropy(error) => write!(f, "webhooks: {error}"),
EndpointError::Database(error) => write!(f, "webhooks: {error}"),
}
}
}
impl std::error::Error for EndpointError {}
impl From<DbError> for EndpointError {
fn from(error: DbError) -> Self {
EndpointError::Database(error)
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum PublishError {
InvalidEventType(String),
Database(DbError),
}
impl std::fmt::Display for PublishError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
PublishError::InvalidEventType(event_type) => {
write!(f, "webhooks: not a usable event type: {event_type:?}")
}
PublishError::Database(error) => write!(f, "webhooks: {error}"),
}
}
}
impl std::error::Error for PublishError {}
impl From<DbError> for PublishError {
fn from(error: DbError) -> Self {
PublishError::Database(error)
}
}
#[derive(Debug, Clone)]
pub struct Published {
pub event_id: String,
pub endpoints: usize,
statements: Vec<cratefield_core::Statement>,
}
impl Published {
#[must_use]
pub fn into_statements(self) -> Vec<cratefield_core::Statement> {
self.statements
}
}
fn is_valid_event_type(event_type: &str) -> bool {
(1..=128).contains(&event_type.len()) && event_type.chars().all(|c| c.is_ascii_graphic())
}
#[must_use]
pub fn matches_filter(filter: &str, event_type: &str) -> bool {
let filter = filter.trim();
filter == "*"
|| filter
.split(',')
.map(str::trim)
.any(|candidate| candidate == event_type)
}
#[must_use]
pub fn normalize_event_types(event_types: &[&str]) -> String {
let mut seen = Vec::new();
for candidate in event_types {
let candidate = candidate.trim();
if !candidate.is_empty() && !seen.contains(&candidate) {
seen.push(candidate);
}
}
if seen.is_empty() {
"*".to_owned()
} else {
seen.join(",")
}
}
fn is_http_url(url: &str) -> bool {
let max = 2048;
(url.starts_with("https://") || url.starts_with("http://"))
&& url.len() <= max
&& !url.contains(char::is_whitespace)
&& match url.split_once("://") {
Some((_, rest)) => {
let authority = rest.split(['/', '?', '#']).next().unwrap_or("");
!authority.is_empty()
}
None => false,
}
}
fn generate_secret() -> Result<String, EndpointError> {
let mut bytes = [0_u8; 32];
getrandom::fill(&mut bytes).map_err(|error| EndpointError::Entropy(error.to_string()))?;
let mut hex = String::with_capacity(2 * bytes.len());
for byte in bytes {
use std::fmt::Write as _;
let _ = write!(hex, "{byte:02x}");
}
Ok(format!("whsec_{hex}"))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn defaults_are_the_documented_ones() {
let module = Webhooks::new();
assert_eq!(module.name(), "webhooks");
assert_eq!(module.version(), env!("CARGO_PKG_VERSION"));
assert_eq!(module.requires(), [Port::Db, Port::HttpClient, Port::Clock]);
assert_eq!(
module.tables(),
[
"webhooks_endpoints",
"webhooks_outbox",
"webhooks_deliveries",
"webhooks_dead_letters"
]
);
assert_eq!(module.max_attempts, 5);
}
#[test]
fn max_attempts_stays_within_the_range_the_config_must_pass() {
assert_eq!(Webhooks::new().max_attempts(0).max_attempts, 1);
assert_eq!(Webhooks::new().max_attempts(2).max_attempts, 2);
assert_eq!(Webhooks::new().max_attempts(1_000).max_attempts, 1_000);
assert_eq!(Webhooks::new().max_attempts(5_000).max_attempts, 1_000);
}
#[test]
fn a_created_endpoints_debug_never_carries_the_secret() {
let created = CreatedEndpoint {
endpoint_id: "ep_01HXYZ".to_owned(),
secret: "whsec_hunter2".to_owned(),
};
let rendered = format!("{created:?}");
assert!(rendered.contains("ep_01HXYZ"), "{rendered}");
assert!(rendered.contains("[redacted]"), "{rendered}");
assert!(!rendered.contains("hunter2"), "{rendered}");
}
#[test]
fn event_types_must_be_header_safe_tokens() {
assert!(is_valid_event_type("order.paid"));
assert!(is_valid_event_type("order-paid_v2"));
assert!(!is_valid_event_type(""), "empty");
assert!(!is_valid_event_type(" order.paid"), "leading space");
assert!(!is_valid_event_type("order paid"), "inner space");
assert!(!is_valid_event_type("command\ninjection"), "control char");
assert!(!is_valid_event_type("héllo"), "non-ASCII");
assert!(is_valid_event_type(&"x".repeat(128)), "the max is allowed");
assert!(!is_valid_event_type(&"x".repeat(129)), "over the max");
}
#[test]
fn a_star_filter_matches_everything_and_a_list_matches_exactly() {
assert!(matches_filter("*", "order.paid"));
assert!(matches_filter("*", ""));
assert!(matches_filter("order.paid", "order.paid"));
assert!(matches_filter(
" order.paid , order.refunded ",
"order.refunded"
));
assert!(!matches_filter("order.paid", "order.paid.v2"));
assert!(!matches_filter("order.paid", "order.refunded"));
assert!(
!matches_filter("", "order.paid"),
"empty stores as *, never as nothing"
);
}
#[test]
fn filters_normalise_to_a_stored_list_or_a_star() {
assert_eq!(
normalize_event_types(&["order.paid", "order.refunded"]),
"order.paid,order.refunded"
);
assert_eq!(normalize_event_types(&[" order.paid ", ""]), "order.paid");
assert_eq!(normalize_event_types(&["a", "a", "b"]), "a,b");
assert_eq!(normalize_event_types(&[]), "*");
assert_eq!(normalize_event_types(&["", " "]), "*");
}
#[test]
fn urls_must_be_http_s_with_nowhere_for_whitespace_to_hide() {
assert!(is_http_url("https://example.com/hook"));
assert!(is_http_url("http://example.com"));
assert!(!is_http_url("example.com/hook"));
assert!(!is_http_url("ftp://example.com"));
assert!(!is_http_url("https://"));
assert!(!is_http_url("http:///path"), "no host at all");
assert!(!is_http_url("https:///q?a=1"), "no host, only a query");
assert!(is_http_url("http://example.com?a=1"), "authority, no path");
assert!(!is_http_url("https://example.com/a b"));
assert!(!is_http_url(""));
assert!(!is_http_url("https://example.com/\n"));
}
#[test]
fn secrets_are_prefixed_hex_and_never_repeat() {
let a = generate_secret().expect("os randomness");
let b = generate_secret().expect("os randomness");
assert!(a.starts_with("whsec_"), "{a}");
let hex = a.strip_prefix("whsec_").expect("prefix");
assert_eq!(hex.len(), 64, "32 bytes in hex");
assert!(
hex.chars()
.all(|c| c.is_ascii_hexdigit() && !c.is_ascii_uppercase())
);
assert_ne!(a, b, "two secrets are two draws");
}
#[test]
fn a_bad_config_is_collected_not_thrown_one_at_a_time() {
let ok = MapConfig::from_pairs([("WEBHOOKS_MAX_ATTEMPTS", "3")]);
assert!(Webhooks::new().validate_config(&ok).is_ok());
let empty = MapConfig::from_pairs([("WEBHOOKS_MAX_ATTEMPTS", "")]);
let error = Webhooks::new()
.validate_config(&empty)
.expect_err("an unparsable attempt count is refused");
assert!(
error.to_string().contains("WEBHOOKS_MAX_ATTEMPTS"),
"{error}"
);
let zero = MapConfig::from_pairs([("WEBHOOKS_MAX_ATTEMPTS", "0")]);
assert!(Webhooks::new().validate_config(&zero).is_err());
let ceiling = MapConfig::from_pairs([("WEBHOOKS_MAX_ATTEMPTS", "1000")]);
assert!(Webhooks::new().validate_config(&ceiling).is_ok());
let past = MapConfig::from_pairs([("WEBHOOKS_MAX_ATTEMPTS", "1001")]);
assert!(Webhooks::new().validate_config(&past).is_err());
}
use cratefield_core::MapConfig;
}