faucet_cli/serve/triggers/
webhook.rs

1//! `POST/PUT /v1/triggers/{name}` — the webhook trigger endpoint. Looks the name
2//! up in the `TriggersHandle` webhook table, checks the method allowlist, applies
3//! a leading-edge debounce (coalesce fires that arrive within `debounce_secs` of
4//! the last accepted fire), builds a `TriggerEvent::Webhook`, and fires. Bearer
5//! auth is inherited from the `/v1` route_layer.
6
7use super::context::TriggerEvent;
8use super::enqueue::{self, FireOutcome};
9use crate::serve::error::ServeError;
10use crate::serve::state::ServerState;
11use crate::serve::triggers::spec::TriggerKind;
12use axum::Json;
13use axum::extract::{Path, RawQuery, State};
14use axum::http::{HeaderMap, Method, StatusCode};
15use serde_json::json;
16use std::collections::BTreeMap;
17
18pub async fn handle(
19    State(state): State<ServerState>,
20    Path(name): Path<String>,
21    method: Method,
22    headers: HeaderMap,
23    RawQuery(raw_query): RawQuery,
24    body: String,
25) -> Result<(StatusCode, Json<serde_json::Value>), ServeError> {
26    let compiled = state
27        .triggers()
28        .webhook(&name)
29        .ok_or(ServeError::NotFound)?;
30
31    let (methods, dedupe_header, debounce_secs) = match &compiled.spec.kind {
32        TriggerKind::Webhook {
33            methods,
34            dedupe_header,
35            debounce_secs,
36        } => (methods, dedupe_header, *debounce_secs),
37        _ => return Err(ServeError::NotFound),
38    };
39    let m = method.as_str().to_ascii_uppercase();
40    if !methods
41        .iter()
42        .any(|allowed| allowed.to_ascii_uppercase() == m)
43    {
44        return Err(ServeError::BadConfig(format!(
45            "method {m} not allowed for webhook trigger '{name}'"
46        )));
47    }
48
49    // Leading-edge debounce: coalesce fires that arrive within `debounce_secs` of
50    // the last accepted fire for this trigger. The first fire (and any after the
51    // window has fully elapsed) is accepted; the rest return `coalesced`.
52    if debounce_secs > 0 {
53        let now_ms = chrono::Utc::now().timestamp_millis();
54        if !state
55            .triggers()
56            .allow_fire(&name, (debounce_secs as i64) * 1000, now_ms)
57        {
58            crate::serve::triggers::metrics::coalesced(&name);
59            return Ok((StatusCode::OK, Json(json!({ "status": "coalesced" }))));
60        }
61    }
62
63    // Idempotency key: dedupe header value, else a per-request UUID.
64    let idem = dedupe_header
65        .as_ref()
66        .and_then(|h| headers.get(h.as_str()))
67        .and_then(|v| v.to_str().ok())
68        .map(|s| s.to_string())
69        .unwrap_or_else(|| uuid::Uuid::new_v4().to_string());
70
71    let header_map: BTreeMap<String, String> = headers
72        .iter()
73        .filter_map(|(k, v)| {
74            v.to_str()
75                .ok()
76                .map(|val| (k.as_str().to_ascii_lowercase(), val.to_string()))
77        })
78        .collect();
79    let query_map = parse_query(raw_query.as_deref());
80
81    let event = TriggerEvent::Webhook {
82        method: m,
83        body,
84        headers: header_map,
85        query: query_map,
86        idem,
87    };
88    let fired_at = chrono::Utc::now().to_rfc3339();
89    match enqueue::fire(&state, &compiled, event, &fired_at).await {
90        FireOutcome::Enqueued(run_id) => {
91            state.triggers().record_ok(&name, Some(fired_at));
92            Ok((
93                StatusCode::ACCEPTED,
94                Json(json!({ "run_id": run_id, "status": "queued" })),
95            ))
96        }
97        FireOutcome::Coalesced => Ok((StatusCode::OK, Json(json!({ "status": "coalesced" })))),
98        FireOutcome::Dropped(reason) => {
99            tracing::warn!(trigger = %name, %reason, "webhook fire dropped");
100            Err(ServeError::QueueFull {
101                retry_after_secs: 5,
102            })
103        }
104        FireOutcome::Error(msg) => {
105            state
106                .triggers()
107                .record_err(&name, msg.clone(), super::watcher::UNHEALTHY_THRESHOLD);
108            Err(ServeError::Internal(msg))
109        }
110    }
111}
112
113fn parse_query(raw: Option<&str>) -> BTreeMap<String, String> {
114    let mut m = BTreeMap::new();
115    if let Some(q) = raw {
116        for pair in q.split('&').filter(|s| !s.is_empty()) {
117            let mut it = pair.splitn(2, '=');
118            let k = it.next().unwrap_or_default().to_string();
119            let v = it.next().unwrap_or_default().to_string();
120            m.insert(k, v);
121        }
122    }
123    m
124}
125
126#[cfg(test)]
127mod tests {
128    use super::*;
129
130    #[test]
131    fn parses_query_pairs() {
132        let q = parse_query(Some("mode=full&tenant=acme"));
133        assert_eq!(q.get("mode").map(String::as_str), Some("full"));
134        assert_eq!(q.get("tenant").map(String::as_str), Some("acme"));
135        assert!(parse_query(None).is_empty());
136    }
137}
faucet_cli/serve/triggers/webhook.rs

faucet_cli/serve/triggers/
webhook.rs
