OXYGENHUB
# 34beb1a7e376 oxygenhub / crates / hub / src / hooks.rs
4.1 KiB · Rust · last changed Milestone 0.5: API, webhooks, checks 1 month ago source blame history raw
1
//! Webhooks (hub spec §12.2): queueing an event for every hook that wants
2
//! it, signing and delivering with retries, and the payload envelope.
3
4
use crate::db::{self, hooks as hdb};
5
use crate::Shared;
6
use hub_api::WebhookEnvelope;
7
8
/// Queue `event` for the hooks of a repository (and its organisation).
9
pub async fn emit(app: &Shared, repo: &db::repos::RepoRow, event: &str, actor: &str, data: serde_json::Value) {
10
    let hooks = match hdb::hooks_hearing(&app.db, repo.id, &repo.owner).await {
11
        Ok(h) => h,
12
        Err(e) => {
13
            tracing::warn!("hooks: {e}");
14
            return;
15
        }
16
    };
17
    if hooks.is_empty() {
18
        return;
19
    }
20
    let repository = hub_api::Repository {
21
        owner: repo.owner.clone(),
22
        name: repo.name.clone(),
23
        description: repo.description.clone(),
24
        default_bookmark: repo.default_bookmark.clone(),
25
        visibility: repo.visibility(),
26
        created_ms: repo.created_ms,
27
        updated_ms: repo.updated_ms,
28
        clone_url: format!("{}/{}", app.config.base_url(), repo.slug()),
29
    };
30
    for h in hooks.into_iter().filter(|h| h.wants(event)) {
31
        let delivery = crate::auth::random_hex(16);
32
        let envelope = WebhookEnvelope {
33
            event: event.into(),
34
            delivery: delivery.clone(),
35
            repository: Some(repository.clone()),
36
            actor: actor.into(),
37
            at_ms: db::now_ms(),
38
            data: data.clone(),
39
        };
40
        let payload = serde_json::to_string(&envelope).unwrap_or_default();
41
        if let Err(e) = hdb::queue_delivery(&app.db, h.id, &delivery, event, &payload).await {
42
            tracing::warn!("queueing delivery: {e}");
43
        }
44
    }
45
    app.events.notify_deliveries();
46
}
47
48
/// `X-Oxyhub-Signature` for a body: keyed BLAKE3 with the hook's secret.
49
pub fn signature(secret: &str, body: &[u8]) -> String {
50
    let key = blake3::hash(secret.as_bytes());
51
    format!("blake3-keyed={}", hex::encode(blake3::keyed_hash(key.as_bytes(), body).as_bytes()))
52
}
53
54
/// Deliver everything that is due; returns how many were attempted.
55
pub async fn deliver_due(app: &Shared) -> usize {
56
    let due = match hdb::due_deliveries(&app.db, 50).await {
57
        Ok(d) => d,
58
        Err(_) => return 0,
59
    };
60
    let client = reqwest::Client::builder()
61
        .timeout(std::time::Duration::from_secs(20))
62
        .user_agent(format!("oxygenhub/{}", env!("CARGO_PKG_VERSION")))
63
        .build();
64
    let Ok(client) = client else { return 0 };
65
    let mut n = 0;
66
    for d in due {
67
        let Ok(Some(hook)) = hdb::hook(&app.db, d.hook_id).await else { continue };
68
        n += 1;
69
        let (status, response) = match client
70
            .post(&hook.url)
71
            .header("content-type", "application/json")
72
            .header("x-oxyhub-event", &d.event)
73
            .header("x-oxyhub-delivery", &d.delivery)
74
            .header("x-oxyhub-signature", signature(&hook.secret, d.payload.as_bytes()))
75
            .body(d.payload.clone())
76
            .send()
77
            .await
78
        {
79
            Ok(r) => {
80
                let status = r.status().as_u16() as i64;
81
                let text = r.text().await.unwrap_or_default();
82
                (Some(status), text)
83
            }
84
            Err(e) => (None, e.to_string()),
85
        };
86
        hdb::record_attempt(&app.db, d.id, d.attempts, status, &response).await.ok();
87
    }
88
    n
89
}
90
91
/// The background worker: delivers when nudged and every minute anyway.
92
pub async fn worker(app: Shared) {
93
    let mut rx = app.events.deliveries.subscribe();
94
    loop {
95
        deliver_due(&app).await;
96
        let cutoff = db::now_ms() - app.config.retention.webhook_log_days as i64 * 86_400_000;
97
        hdb::prune_deliveries(&app.db, cutoff).await.ok();
98
        tokio::select! {
99
            _ = rx.recv() => {}
100
            _ = tokio::time::sleep(std::time::Duration::from_secs(60)) => {}
101
        }
102
    }
103
}
104
105
#[cfg(test)]
106
mod tests {
107
    use super::*;
108
109
    #[test]
110
    fn signatures_are_keyed() {
111
        let a = signature("s1", b"body");
112
        assert!(a.starts_with("blake3-keyed="));
113
        assert_ne!(a, signature("s2", b"body"));
114
        assert_ne!(a, signature("s1", b"other"));
115
    }
116
}