webhooksmith

Webhook delivery for Rust backed by Postgres or SQLite. Atomic outbox writes, HMAC-SHA256 signing, automatic retry with exponential backoff, and dead letter queue. No external services.

Postgres backend (default — best for production, transactional outbox):

[dependencies]
webhooksmith = "0.1"
tokio = { version = "1", features = ["full"] }
serde_json = "1"

SQLite backend (desktop apps, CLI tools, embedded — no Postgres needed):

[dependencies]
webhooksmith = { version = "0.1", features = ["sqlite"] }
tokio = { version = "1", features = ["full"] }
serde_json = "1"

With SQLite, use SqliteEngine — same API as WebhookEngine:

use webhooksmith::SqliteEngine;

let engine = SqliteEngine::new("sqlite:webhooks.db").await?;
engine.migrate().await?;
// All the same methods: send, broadcast, retry_dead, queue_stats, etc.

How it works

1. Your app registers partner webhook endpoints in Postgres or SQLite.
2. When an event happens, you call engine.send() — the event is saved to your database.
3. The background worker picks it up and POSTs it with an HMAC-SHA256 signature.
4. Failures retry with exponential backoff. After max_attempts failures, the event moves to a dead-letter queue.

No Redis, no queuing service, no external infrastructure. Just your existing database.

Quick start

use webhooksmith::WebhookEngine;
use serde_json::json;
use std::time::Duration;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let engine = WebhookEngine::builder()
        .database_url("postgres://user:pass@localhost/mydb")
        .build()
        .await?;

    // Creates the 3 required tables (safe to call on every startup)
    engine.migrate().await?;

    // Register a partner endpoint
    let endpoint = engine
        .register("https://partner.example.com/webhooks", "your-signing-secret-32chars")
        .await?;

    // Queue a webhook event
    engine.send("order.created", json!({"id": 1001, "total": 49.99}), endpoint.id).await?;

    // Start the delivery worker (blocks until the process exits)
    engine.run().await;
}

Transactional outbox

Write your business data and the webhook event in the same database transaction. If the transaction rolls back, the event never exists. If it commits, the event is guaranteed to be delivered.

let mut tx = engine.pool().begin().await?;

// Your business logic
sqlx::query!("INSERT INTO orders (id, total) VALUES ($1, $2)", order_id, 49.99)
    .execute(&mut *tx)
    .await?;

// Webhook in the same transaction — only queued if this tx commits
engine
    .send_in_tx("order.created", json!({"id": order_id}), endpoint.id, &mut tx)
    .await?;

tx.commit().await?;

Fan-out to all endpoints at once, atomically:

let mut tx = engine.pool().begin().await?;
engine.broadcast_in_tx("order.created", json!({"id": 1001}), &mut tx).await?;
tx.commit().await?;

All sending methods

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",
}

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.

// 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

// 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

// Register
let ep = engine.register("https://partner.com/hooks", "secret-32chars-min").await?;

// Register with full config
let ep = engine.register_with(webhooksmith::NewEndpoint {
    url: "https://partner.com/hooks".into(),
    signing_secret: "secret-32chars-min".into(),
    description: Some("Partner A".into()),
    max_attempts: Some(10),       // retries before DLQ (default: 10, min: 1)
    initial_delay_ms: Some(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,
    created_at: "2026-01-01T00:00:00Z",
    updated_at: "2026-01-01T00:00:00Z",  // updated automatically by DB trigger
}

Monitoring & queue operations

// 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?;

Dead letter queue

When an event exceeds max_attempts failures it moves to status = Dead.

// 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.

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:

// Reset events stuck in 'delivering' for > 120 seconds (returns count reset)
let reset = engine.recover_stuck_deliveries(Duration::from_secs(120)).await?;

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 on the receiving side with webhooksmith-axum:

use axum::{Router, routing::post, http::StatusCode};
use webhooksmith_axum::{WebhookSecretLayer, VerifiedWebhook};

async fn handle(VerifiedWebhook(payload): VerifiedWebhook) -> StatusCode {
    println!("{}: {:?}", payload.event_type, payload.body);
    StatusCode::OK
}

let app: Router = Router::new()
    .route("/webhooks", post(handle))
    .layer(WebhookSecretLayer::new("your-signing-secret"));

Verifying manually without the axum crate:

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:

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

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

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:

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

# Postgres examples need Docker:
docker compose up -d

# 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:.

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)

License

MIT OR Apache-2.0