webhooksmith 0.1.3

Webhook delivery for Rust — atomic outbox pattern, HMAC-SHA256 signing, retry with backoff, dead letter queue. Supports Postgres and SQLite.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217

//! End-to-end tests for the SQLite backend.
//! Uses in-memory SQLite — no external DB needed.

#[cfg(feature = "sqlite")]
mod sqlite_tests {
    use webhooksmith::{EventStatus, SqliteEngine};
    use serde_json::json;
    use std::time::Duration;
    use wiremock::{matchers::method, Mock, MockServer, ResponseTemplate};

    async fn engine() -> SqliteEngine {
        let e = SqliteEngine::builder()
            .database_url("sqlite::memory:")
            .allow_insecure_urls()
            .build()
            .await
            .unwrap();
        e.migrate().await.unwrap();
        e
    }

    #[tokio::test]
    async fn sqlite_register_and_send() {
        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .respond_with(ResponseTemplate::new(200))
            .mount(&server)
            .await;

        let e = engine().await;
        let ep = e.register(&format!("{}/hook", server.uri()), "sqlite_test_secret_32chars_ok").await.unwrap();

        let event = e.send("order.created", json!({"id": 1}), ep.id).await.unwrap();
        assert_eq!(event.status, EventStatus::Pending);

        let n = e.run_once().await.unwrap();
        assert_eq!(n, 1);

        let after = e.event(event.id).await.unwrap().unwrap();
        assert_eq!(after.status, EventStatus::Delivered);
    }

    #[tokio::test]
    async fn sqlite_send_delivers_event() {
        // SQLite limitation: true transactional outbox (send_in_tx) is not supported
        // because max_connections=1 — the transaction holds the only connection and
        // a second operation would deadlock. Use the Postgres backend for transactional outbox.
        // This test verifies basic send + deliver works correctly.
        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .respond_with(ResponseTemplate::new(200))
            .mount(&server)
            .await;

        let e = engine().await;
        let ep = e.register(&format!("{}/hook", server.uri()), "sqlite_test_secret_32chars_ok").await.unwrap();

        let ev = e.send("order.shipped", json!({"id": 2}), ep.id).await.unwrap();
        assert!(e.event(ev.id).await.unwrap().is_some(), "event must exist after send");

        e.run_once().await.unwrap();
        let after = e.event(ev.id).await.unwrap().unwrap();
        assert_eq!(after.status, EventStatus::Delivered);
    }

    #[tokio::test]
    async fn sqlite_idempotency() {
        let e = engine().await;
        let server = MockServer::start().await;
        let ep = e.register(&format!("{}/hook", server.uri()), "sqlite_test_secret_32chars_ok").await.unwrap();

        let e1 = e.send_idempotent("order.created", json!({}), ep.id, "key-1").await.unwrap();
        let e2 = e.send_idempotent("order.created", json!({}), ep.id, "key-1").await.unwrap();
        assert_eq!(e1.id, e2.id, "same key must return same event");

        let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM webhook_events")
            .fetch_one(e.pool()).await.unwrap();
        assert_eq!(count, 1);
    }

    #[tokio::test]
    async fn sqlite_broadcast() {
        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .respond_with(ResponseTemplate::new(200))
            .mount(&server)
            .await;

        let e = engine().await;
        for _ in 0..3 {
            e.register(&format!("{}/hook", server.uri()), "sqlite_test_secret_32chars_ok").await.unwrap();
        }

        let events = e.broadcast("announcement", json!({"msg": "hello"})).await.unwrap();
        assert_eq!(events.len(), 3);
    }

    #[tokio::test]
    async fn sqlite_queue_stats() {
        let e = engine().await;
        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .respond_with(ResponseTemplate::new(500)) // all fail
            .mount(&server)
            .await;

        let ep = e.register_with(webhooksmith::NewEndpoint {
            url: format!("{}/hook", server.uri()),
            signing_secret: "sqlite_test_secret_32chars_ok".into(),
            description: None,
            max_attempts: Some(1),
            initial_delay_ms: Some(1),
            event_filter: None,
        }).await.unwrap();

        for i in 0..3 { e.send("test", json!({"i": i}), ep.id).await.unwrap(); }

        let stats = e.queue_stats().await.unwrap();
        assert_eq!(stats.pending, 3);

        e.run_once().await.unwrap(); // all fail → dead (max_attempts=1)
        let stats2 = e.queue_stats().await.unwrap();
        assert_eq!(stats2.dead, 3);
    }

    #[tokio::test]
    async fn sqlite_retry_dead() {
        let e = engine().await;
        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .respond_with(ResponseTemplate::new(500))
            .up_to_n_times(1)
            .mount(&server)
            .await;
        Mock::given(method("POST"))
            .respond_with(ResponseTemplate::new(200))
            .mount(&server)
            .await;

        // Register with max_attempts=1 so first failure goes to DLQ
        let ep = e.register_with(webhooksmith::NewEndpoint {
            url: format!("{}/hook", server.uri()),
            signing_secret: "sqlite_test_secret_32chars_ok".into(),
            description: None,
            max_attempts: Some(1),
            initial_delay_ms: Some(1),
            event_filter: None,
        }).await.unwrap();

        let ev = e.send("test", json!({}), ep.id).await.unwrap();
        e.run_once().await.unwrap();

        let dead = e.event(ev.id).await.unwrap().unwrap();
        assert_eq!(dead.status, EventStatus::Dead);

        e.retry_dead(ev.id).await.unwrap();
        e.run_once().await.unwrap();
        let final_state = e.event(ev.id).await.unwrap().unwrap();
        assert_eq!(final_state.status, EventStatus::Delivered);
    }

    #[tokio::test]
    async fn sqlite_cleanup() {
        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .respond_with(ResponseTemplate::new(200))
            .mount(&server)
            .await;

        let e = engine().await;
        let ep = e.register(&format!("{}/hook", server.uri()), "sqlite_test_secret_32chars_ok").await.unwrap();

        for i in 0..5 { e.send("test", json!({"i": i}), ep.id).await.unwrap(); }
        e.run_once().await.unwrap(); // all delivered

        // Backdate
        sqlx::query("UPDATE webhook_events SET created_at = datetime('now', '-2 days') WHERE status='delivered'")
            .execute(e.pool()).await.unwrap();

        let removed = e.cleanup_delivered(Duration::from_secs(86_400)).await.unwrap();
        assert_eq!(removed, 5);

        let stats = e.queue_stats().await.unwrap();
        assert_eq!(stats.delivered, 0);
    }

    #[tokio::test]
    async fn sqlite_list_endpoints() {
        let e = engine().await;
        assert_eq!(e.list_endpoints().await.unwrap().len(), 0);

        e.register("https://a.example.com/hook", "sqlite_test_secret_32chars_ok").await.unwrap();
        e.register("https://b.example.com/hook", "sqlite_test_secret_32chars_ok").await.unwrap();

        assert_eq!(e.list_endpoints().await.unwrap().len(), 2);
    }

    #[tokio::test]
    async fn sqlite_disable_endpoint_stops_delivery() {
        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .respond_with(ResponseTemplate::new(200))
            .expect(0)
            .mount(&server)
            .await;

        let e = engine().await;
        let ep = e.register(&format!("{}/hook", server.uri()), "sqlite_test_secret_32chars_ok").await.unwrap();

        e.disable_endpoint(ep.id).await.unwrap();
        e.send("test", json!({}), ep.id).await.unwrap();
        let n = e.run_once().await.unwrap();
        assert_eq!(n, 0, "disabled endpoint must not be claimed");

        server.verify().await;
    }
}