# webhooksmith
[](https://crates.io/crates/webhooksmith)
[](https://docs.rs/webhooksmith)
[](https://github.com/biplabku/webhooksmith/actions/workflows/ci.yml)
[](LICENSE-MIT)
Webhook delivery for Rust backed by Postgres or SQLite. HMAC-SHA256 signing,
automatic retry with exponential backoff, dead letter queue, and event type filtering.
Postgres backend: full transactional outbox (atomic writes). SQLite backend: persistent delivery without Postgres.
---
## Crates
| [`webhooksmith`](webhooksmith/) | Core engine — sending, delivery worker, DLQ, monitoring. Supports Postgres and SQLite. |
| [`webhooksmith-axum`](webhooksmith-axum/) | Axum extractor for verifying incoming webhooks + admin HTTP router |
| [`webhooksmith-actix`](webhooksmith-actix/) | Actix-web extractor for verifying incoming webhooks |
---
## How it works
1. Register partner webhook endpoints in Postgres or SQLite.
2. Call `engine.send()` — the event is persisted atomically.
3. The background worker picks it up and POSTs it with an HMAC-SHA256 signature.
4. On failure: exponential backoff → dead letter queue → manual retry.
No Redis. No queuing service. Just your existing database.
```mermaid
flowchart LR
A[Webhook event created<br/>engine.send / send_in_tx] --> B[(Outbox table<br/>transactional write)]
B --> C[Background dispatcher<br/>engine.run]
C --> D{HTTP delivery attempt<br/>HMAC-SHA256 signed}
D -->|2xx response| E[Delivered]
D -->|failure or timeout| F{Retries exhausted?}
F -->|no| G[Exponential backoff<br/>with jitter]
G --> C
F -->|yes| H[[Dead letter queue]]
H -->|manual retry_dead| C
```
---
## Quick start
**Postgres** (default — best for production, transactional outbox):
```toml
[dependencies]
webhooksmith = "0.2"
tokio = { version = "1", features = ["full"] }
serde_json = "1"
```
**SQLite** (desktop apps, CLI tools, embedded, no Postgres needed):
```toml
[dependencies]
webhooksmith = { version = "0.2", features = ["sqlite"] }
tokio = { version = "1", features = ["full"] }
serde_json = "1"
```
With SQLite, replace `WebhookEngine` with `SqliteEngine` — same API:
```rust
use webhooksmith::SqliteEngine;
let engine = SqliteEngine::new("sqlite:webhooks.db").await?;
engine.migrate().await?;
```
> **SQLite transactional outbox:** `SqliteEngine::send_in_tx()` provides true atomicity.
> The event is written on the transaction's connection — rollback removes it, commit persists it.
> Both backends support the full transactional outbox pattern.
```rust
use webhooksmith::WebhookEngine;
use serde_json::json;
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let engine = WebhookEngine::builder()
.database_url("postgres://user:pass@localhost/mydb")
.build()
.await?;
engine.migrate().await?; // creates 3 tables, safe on every startup
let endpoint = engine
.register("https://partner.example.com/webhooks", "your-signing-secret-32chars")
.await?;
// Send a webhook
engine.send("order.created", json!({"id": 1001}), endpoint.id).await?;
// Start background delivery worker
engine.run().await;
}
```
---
## Transactional outbox
Write business data and webhook event in the same transaction — neither exists without the other:
```rust
let mut tx = engine.pool().begin().await?;
sqlx::query!("INSERT INTO orders (id) VALUES ($1)", order_id)
.execute(&mut *tx).await?;
engine.send_in_tx("order.created", json!({"id": order_id}), endpoint.id, &mut tx).await?;
tx.commit().await?; // webhook only queued if this succeeds
```
Fan-out to all endpoints at once, atomically:
```rust
let mut tx = engine.pool().begin().await?;
engine.broadcast_in_tx("order.created", json!({"id": 1001}), &mut tx).await?;
tx.commit().await?;
```
---
## Key properties
| **Transactional outbox** | Business data and webhook event written atomically — no phantom events, no silent drops (Postgres + SQLite) |
| **Circuit breaker** | After 5 consecutive failures, endpoint paused with exponential backoff (5→10→20 min, capped at 320) |
| **Idempotency keys** | `send_idempotent("order.created", payload, endpoint_id, "order-1001")` — safe to retry, never duplicates |
| **Event type filtering** | Endpoints subscribe to `"order.*"` or `"payment.captured"` — `broadcast()` routes automatically |
| **Admin HTTP API** | `webhooksmith_axum::admin(engine)` — stats, endpoint list, DLQ inspection, bulk retry |
| **Graceful shutdown** | `run_graceful(ctrl_c_signal)` — current batch drains before exit (Kubernetes-safe) |
| **SSRF protection** | Private IPs, loopback, link-local blocked at registration AND at delivery (DNS rebinding protection) |
| **HMAC-SHA256 signing** | Constant-time verification, replay protection, Svix-compatible format |
| **Dead letter queue** | Events that exhaust retries are inspectable and requeueable, not silently dropped |
| **Multi-backend** | Postgres (full transactional outbox) or SQLite (no Postgres required) |
| **Multi-framework receiver** | Verify incoming webhooks in axum (`webhooksmith-axum`) or actix-web (`webhooksmith-actix`) |
---
## All APIs at a glance
**Sending:**
[`send`](#all-sending-methods) · [`send_in_tx`](#all-sending-methods) · [`send_idempotent`](#idempotency-keys) · [`send_idempotent_in_tx`](#idempotency-keys)
[`broadcast`](#all-sending-methods) · [`broadcast_in_tx`](#all-sending-methods) · [`broadcast_idempotent`](#idempotency-keys)
**Endpoints:**
[`register`](#endpoint-management) · [`register_with`](#endpoint-management) · [`endpoint`](#endpoint-management) · [`list_endpoints`](#endpoint-management) · [`list_endpoints_paged`](#endpoint-management)
[`update_endpoint`](#endpoint-management) · [`enable_endpoint`](#endpoint-management) · [`disable_endpoint`](#endpoint-management) · [`delete_endpoint`](#endpoint-management)
[`set_event_filter`](#event-type-filtering) · [`clear_event_filter`](#event-type-filtering)
**Worker:**
[`run`](#running-the-worker) · [`run_graceful`](#running-the-worker) · [`run_once`](#running-the-worker)
**Events & monitoring:**
[`event`](#monitoring--queue-operations) · [`events_by_status`](#monitoring--queue-operations) · [`events_global`](#monitoring--queue-operations) · [`queue_stats`](#monitoring--queue-operations) · [`delivery_log`](#monitoring--queue-operations)
**DLQ:**
[`dead_events`](#dead-letter-queue) · [`dead_events_paged`](#dead-letter-queue) · [`retry_dead`](#dead-letter-queue) · [`retry_all_dead`](#dead-letter-queue)
**Cleanup & ops:**
[`cleanup_delivered`](#cleanup) · [`cleanup_dead`](#cleanup) · [`recover_stuck_deliveries`](#crash-recovery)
---
## All sending methods
| `send(event_type, payload, endpoint_id)` | Send to one endpoint |
| `send_in_tx(event_type, payload, endpoint_id, &mut tx)` | Send to one endpoint, inside your transaction |
| `send_idempotent(event_type, payload, endpoint_id, key)` | Send, deduplicated by key — safe to retry |
| `send_idempotent_in_tx(event_type, payload, endpoint_id, key, &mut tx)` | Idempotent + transactional |
| `broadcast(event_type, payload)` | Fan-out to ALL enabled endpoints |
| `broadcast_in_tx(event_type, payload, &mut tx)` | Fan-out inside your transaction |
| `broadcast_idempotent(event_type, payload, key)` | Fan-out, deduplicated per endpoint by key |
**Validation rules:**
- `event_type` must be non-empty, non-whitespace, ≤ 256 bytes, no control characters
- `payload` must be valid JSON, ≤ 1 MB
**Return value (`WebhookEvent`):**
```
WebhookEvent {
id: Uuid, // unique event ID
endpoint_id: Uuid, // which endpoint this is for
event_type: "order.created",
payload: {"id": 1001, "total": 49.99},
status: Pending, // Pending | Delivering | Delivered | Failed | Dead
attempts: 0,
scheduled_at: "2026-01-01T00:00:00Z",
delivering_since: None,
idempotency_key: None,
created_at: "2026-01-01T00:00:00Z",
}
```
---
## Event type filtering
Each endpoint can subscribe to specific event types. `broadcast()` routes automatically — only matching endpoints receive the event.
```rust
// Subscribe to order events only, using the NewEndpoint builder
engine.register_with(
webhooksmith::NewEndpoint::new("https://orders.example.com/hook", "your-signing-secret-32chars")
.events(["order.*"]),
).await?;
// broadcast() routes — payment.captured does NOT go to order-only endpoints
engine.broadcast("order.created", payload).await?;
// Update filter at any time
engine.set_event_filter(ep_id, vec!["order.*".into(), "payment.captured".into()]).await?;
engine.clear_event_filter(ep_id).await?; // back to receiving all
```
**Pattern rules:**
| `"order.created"` | Exactly `"order.created"` |
| `"order.*"` | Any event starting with `"order."` (e.g. `"order.created"`, `"order.cancelled"`) |
| `"*"` | Everything |
| `None` (no filter) | Everything — default, backward compatible |
| `Some(vec![])` (empty) | Nothing |
`send()` to a specific endpoint always delivers — the filter only applies to `broadcast()`.
---
## Idempotency keys
Protect against double-sends when your code retries on network errors.
The same key for the same endpoint always returns the same event, no matter how many times you call it.
```rust
// First call: creates the event
let ev1 = engine.send_idempotent("order.created", payload.clone(), ep.id, "order-1001").await?;
// Second call (e.g. after a retry): returns the SAME event
let ev2 = engine.send_idempotent("order.created", payload.clone(), ep.id, "order-1001").await?;
assert_eq!(ev1.id, ev2.id); // same event, not a duplicate
```
Key is scoped per `(endpoint_id, key)`. The same key is independent across different endpoints,
which makes `broadcast_idempotent` safe to call multiple times.
---
## Running the worker
```rust
// Blocks forever — put this at the end of main()
engine.run().await;
// Graceful shutdown — current batch drains before exit
engine.run_graceful(async {
tokio::signal::ctrl_c().await.ok();
}).await;
// One cycle — useful for testing or cron-style invocation
let delivered_count: usize = engine.run_once().await?;
```
**What happens in each cycle:**
1. Reset events stuck in `delivering` state for longer than `stuck_timeout` (crash recovery)
2. Claim a batch of due events with `SELECT FOR UPDATE SKIP LOCKED`
3. Deliver each event concurrently via HTTP POST with HMAC signature
4. Record success or failure; schedule retry or move to DLQ
---
## Endpoint management
```rust
// Register
let ep = engine.register("https://partner.com/hooks", "secret-32chars-min").await?;
// Register with full config, using the NewEndpoint builder
let ep = engine.register_with(
webhooksmith::NewEndpoint::new("https://partner.com/hooks", "secret-32chars-min")
.description("Partner A")
.max_attempts(10) // retries before DLQ (default: 10, min: 1)
.initial_delay_ms(1000), // first retry delay ms (default: 1000, min: 1)
).await?;
// Fetch one
let ep: Option<Endpoint> = engine.endpoint(ep.id).await?;
// List all (ordered by created_at)
let all: Vec<Endpoint> = engine.list_endpoints().await?;
// Paginated list
let page: Vec<Endpoint> = engine.list_endpoints_paged(20, 0).await?; // limit=20, offset=0
// Update (only fields you set are changed)
let updated = engine.update_endpoint(ep.id, webhooksmith::UpdateEndpoint {
url: Some("https://new-partner.com/hooks".into()),
max_attempts: Some(5),
..Default::default()
}).await?;
// Enable / disable (stops the worker from delivering to this endpoint)
engine.disable_endpoint(ep.id).await?;
engine.enable_endpoint(ep.id).await?;
// Delete (cascade-deletes all events for this endpoint)
engine.delete_endpoint(ep.id).await?;
```
**Validation rules:**
- `url` must be `http://` or `https://` and not point to a private/loopback/link-local address
- `signing_secret` must be ≥ 16 characters
- `max_attempts` must be ≥ 1
- `initial_delay_ms` must be ≥ 1
**`Endpoint` fields:**
```
Endpoint {
id: Uuid,
url: "https://partner.com/hooks",
signing_secret: "your-secret",
description: Some("Partner A"),
enabled: true,
max_attempts: 10,
initial_delay_ms: 1000,
event_filter: Some(["order.*"]), // None = receive all
consecutive_failures: 0, // circuit breaker counter
circuit_open_until: None, // Some(DateTime) when circuit is open
created_at: "2026-01-01T00:00:00Z",
updated_at: "2026-01-01T00:00:00Z",
}
```
---
## Monitoring & queue operations
```rust
// Count events by status — one DB query
let stats: QueueStats = engine.queue_stats().await?;
// QueueStats { pending: 42, delivering: 3, failed: 1, dead: 0, delivered: 1500 }
// Events for one endpoint with a specific status (paginated, newest first)
let failed: Vec<WebhookEvent> = engine
.events_by_status(ep.id, EventStatus::Failed, 20, 0)
.await?;
// Events across ALL endpoints (for global monitoring dashboards)
let all_failed: Vec<WebhookEvent> = engine
.events_global(EventStatus::Failed, 50, 0)
.await?;
// Full delivery attempt history for one event
let log: Vec<DeliveryAttempt> = engine.delivery_log(event.id).await?;
// DeliveryAttempt { attempted_at, response_status: Some(500), duration_ms: Some(243), error: Some("HTTP 500"), success: false }
// Get one event by ID
let ev: Option<WebhookEvent> = engine.event(event_id).await?;
```
---
## Circuit breaker
After 5 consecutive delivery failures the endpoint's circuit opens. The worker skips it
until the open window expires (5 min → 10 min → 20 min … capped at 320 min, doubling each time).
A single successful delivery resets the counter. Manual DLQ retry (`retry_all_dead`) also resets it.
```rust
let ep = engine.endpoint(ep_id).await?.unwrap();
println!("failures: {}", ep.consecutive_failures);
println!("open until: {:?}", ep.circuit_open_until); // None = closed
```
---
## Dead letter queue
When an event exceeds `max_attempts` failures it moves to `status = Dead`.
```rust
// List dead events for one endpoint (all, unordered)
let dead: Vec<WebhookEvent> = engine.dead_events(ep.id).await?;
// Paginated (newest first)
let page: Vec<WebhookEvent> = engine.dead_events_paged(ep.id, 20, 0).await?;
// Requeue one event (resets attempts to 0)
engine.retry_dead(event.id).await?;
// Returns HooksmithError::InvalidState if event is not dead
// Returns HooksmithError::EventNotFound if event doesn't exist
// Requeue all dead events for one endpoint — returns count requeued
let requeued: u64 = engine.retry_all_dead(ep.id).await?;
```
---
## Cleanup
Delivered and dead events accumulate indefinitely unless cleaned up.
```rust
use std::time::Duration;
// Delete delivered events older than 7 days — returns count deleted
let removed = engine.cleanup_delivered(Duration::from_secs(7 * 86_400)).await?;
// Delete dead events older than 30 days — returns count deleted
let removed = engine.cleanup_dead(Duration::from_secs(30 * 86_400)).await?;
```
Only the specified status is touched — pending, delivering, and failed events are never deleted by cleanup.
---
## Crash recovery
The worker calls this automatically every cycle. You can also call it manually for ops use:
```rust
// Reset events stuck in 'delivering' for > 120 seconds (returns count reset)
let reset = engine.recover_stuck_deliveries(Duration::from_secs(120)).await?;
```
---
## Receiving webhooks
### axum
```toml
[dependencies]
webhooksmith-axum = "0.2"
```
```rust
use axum::{Router, routing::post, http::StatusCode};
use webhooksmith_axum::{WebhookSecretLayer, VerifiedWebhook, TypedWebhook};
use serde::Deserialize;
#[derive(Deserialize)]
struct OrderCreated { order_id: u64 }
// VerifiedWebhook gives you the raw verified payload
async fn handle_raw(VerifiedWebhook(body): VerifiedWebhook) -> StatusCode {
println!("{}: {:?}", body.event_type, body.body);
StatusCode::OK
}
// TypedWebhook<T> deserializes the body for you
async fn handle_typed(TypedWebhook(order): TypedWebhook<OrderCreated>) -> StatusCode {
println!("order: {}", order.order_id);
StatusCode::OK
}
let app: Router = Router::new()
.route("/webhooks", post(handle_raw))
.route("/orders", post(handle_typed))
.layer(WebhookSecretLayer::new("your-signing-secret"));
```
→ [webhooksmith-axum README](webhooksmith-axum/README.md)
### actix-web
```toml
[dependencies]
webhooksmith-actix = "0.2"
```
```rust
use actix_web::{web, App, HttpResponse, Responder};
use webhooksmith_actix::{WebhookSecret, TypedWebhook};
use serde::Deserialize;
#[derive(Deserialize)]
struct OrderCreated { order_id: u64 }
async fn handle(webhook: TypedWebhook<OrderCreated>) -> impl Responder {
println!("order: {}", webhook.payload.order_id);
HttpResponse::Ok().finish()
}
App::new()
.app_data(WebhookSecret::new("your-signing-secret"))
.route("/webhooks", web::post().to(handle))
```
→ [webhooksmith-actix README](webhooksmith-actix/README.md)
### Admin panel (axum)
```rust
use webhooksmith_axum::admin;
let app = Router::new()
.nest("/admin", admin(Arc::clone(&engine)));
// GET /admin/stats · GET /admin/endpoints · GET /admin/dlq/:endpoint_id
// POST /admin/dlq/:endpoint_id/retry-all · GET /admin/metrics (Prometheus)
```
→ [Admin API docs](docs/03-admin-api.md)
---
## What is sent to the endpoint
The worker sends an HTTP POST with:
```
POST https://partner.example.com/webhooks
Content-Type: application/json
x-hooksmith-signature: v1,<hex-encoded-hmac-sha256>
x-hooksmith-timestamp: 1735689600
x-hooksmith-event-id: 550e8400-e29b-41d4-a716-446655440000
x-hooksmith-event-type: order.created
{"id": 1001, "total": 49.99}
```
**Signature format** (Svix-compatible):
The HMAC-SHA256 is computed over `{timestamp}.{body}` using your signing secret.
**Verifying manually** without the axum or actix-web crates:
```rust
use webhooksmith::signing;
fn verify_request(secret: &str, timestamp_header: &str, signature_header: &str, body: &[u8]) -> bool {
let timestamp: i64 = timestamp_header.parse().unwrap_or(0);
signing::verify(secret, timestamp, body, signature_header)
// Returns false if: wrong secret, timestamp older than 5 minutes, or tampered body
}
```
---
## Retry behaviour
Failed deliveries are retried with exponential backoff and full jitter:
| 1st retry | `initial_delay_ms` × 2 (default: 2s) |
| 2nd retry | `initial_delay_ms` × 4 (default: 4s) |
| … | … |
| Any | Capped at 1 hour |
After `max_attempts` failures (default: 10), the event moves to `status = Dead` and is not retried again automatically. Use `retry_dead` or `retry_all_dead` to requeue.
**HTTP responses treated as failure:** anything outside 2xx. `3xx` redirects are explicitly not followed.
---
## Configuration
```rust
let engine = WebhookEngine::builder()
.database_url("postgres://user:pass@localhost/mydb")
// OR: .pool(existing_sqlx_pool)
// Worker
.batch_size(50) // events per cycle (default: 50, min: 1)
.poll_interval(Duration::from_millis(500)) // idle sleep (default: 500ms)
.http_timeout(Duration::from_secs(30)) // per-request timeout (default: 30s)
.stuck_timeout(Duration::from_secs(120)) // reaper threshold (default: 120s)
// http_timeout MUST be < stuck_timeout (panics otherwise)
// Postgres pool
.max_connections(20) // pool size (default: 20)
.acquire_timeout(Duration::from_secs(10)) // connection wait limit (default: 10s)
// Development only — skips SSRF URL validation
// .allow_insecure_urls()
.build()
.await?;
```
---
## SSRF protection
The following URL targets are blocked at endpoint registration and at delivery time (DNS rebinding protection):
- Loopback: `127.x.x.x`, `::1`, `localhost`
- Private IPv4: `10.x`, `172.16–31.x`, `192.168.x`
- Link-local: `169.254.x.x` (AWS metadata endpoint), `fe80::/10`
- CGNAT: `100.64.0.0/10`
- IPv6 unique local: `fc00::/7`
- IPv4-mapped IPv6: `::ffff:10.x.x.x`, etc.
HTTP redirects are blocked at delivery time (the redirect target is not visited).
---
## Error types
```rust
use webhooksmith::HooksmithError;
match result {
Err(HooksmithError::Database(e)) => // sqlx database error
Err(HooksmithError::Http(e)) => // reqwest HTTP error
Err(HooksmithError::EndpointNotFound(id)) => // endpoint doesn't exist
Err(HooksmithError::EventNotFound(id)) => // event doesn't exist
Err(HooksmithError::InvalidState(id)) => // operation invalid for current state
Err(HooksmithError::PayloadTooLarge(n, m)) => // payload n bytes exceeds m byte limit
Err(HooksmithError::Config(msg)) => // validation error (URL, secret, etc.)
Err(HooksmithError::Signing(msg)) => // HMAC error
}
```
---
## Database schema
`engine.migrate()` creates three tables:
```sql
webhook_endpoints -- registered delivery targets
webhook_events -- outbound events, one row per (endpoint, event)
webhook_delivery_attempts -- log of every HTTP call made
```
All tables use UUIDs as primary keys. Postgres uses `TIMESTAMPTZ`; SQLite stores timestamps as ISO 8601 TEXT.
The `webhook_events` table has a partial index on `(status, scheduled_at)` for efficient worker queries.
---
## How-to examples
```bash
docker compose up -d # start Postgres
# Minimal setup — connect, register, send, deliver
cargo run --example basic -p webhooksmith
# Transactional outbox — atomic writes, rollback safety
cargo run --example outbox -p webhooksmith
# Full axum integration — engine in State, graceful shutdown
cargo run --example axum_integration -p webhooksmith
# Multi-tenant — per-customer endpoints, broadcast, idempotent sends
cargo run --example multi_tenant -p webhooksmith
# Monitoring — queue stats, DLQ alerts, retry failed events, cleanup
cargo run --example monitoring -p webhooksmith
# Full end-to-end demo with real axum receiver
cargo run --example demo -p webhooksmith
```
**SQLite examples need no setup** — just pass a file path or `sqlite::memory:`.
---
## How-to guides
| [01 — Transactional outbox](docs/01-outbox-pattern.md) | Atomic business write + webhook event |
| [02 — Circuit breaker](docs/02-circuit-breaker.md) | Per-endpoint failure protection |
| [03 — Admin API](docs/03-admin-api.md) | Stats, DLQ, bulk retry via HTTP |
| [04 — Receiving webhooks](docs/04-receiving-webhooks.md) | axum + actix-web integrations |
| [05 — Event filtering](docs/05-event-filtering.md) | Per-endpoint event subscriptions |
| [06 — DLQ and monitoring](docs/06-dlq-and-monitoring.md) | Dead letter queue, cleanup, delivery log |
| [07 — Graceful shutdown](docs/07-graceful-shutdown.md) | Kubernetes-safe worker lifecycle |
---
## Running the tests
```bash
docker compose up -d
DATABASE_URL=postgres://webhooksmith:webhooksmith@localhost:5432/webhooksmith cargo test --features sqlite
```
276 tests covering unit, integration (real Postgres + real HTTP + in-memory SQLite), true end-to-end (real actix-web receiver over TCP), circuit breaker, admin API, adversarial, stress, bombardment, and event filter routing.
---
## Requirements
- Rust 1.75+
- **Postgres backend (default):** Postgres 14+ (`gen_random_uuid()`, `FOR UPDATE SKIP LOCKED`, partial unique indexes, triggers)
- **SQLite backend:** SQLite 3.35+ (supports `RETURNING`; WAL mode enabled automatically)
---
See [CHANGELOG.md](CHANGELOG.md) for release history.
## License
MIT OR Apache-2.0