pagesift_api/webhook/
mod.rs

1//! Firecrawl v2 的任务 webhook：`started`、`page`、`completed`、`failed` 四种事件。
2//!
3//! - 每个任务一个投递任务，事件按产生顺序逐个 POST，失败按指数退避重试，次数有上限；
4//! - 每个事件带 `eventId`（`{任务 id}:{seq}`）和 `seq`（任务内从 1 起单调递增），接收方据此去重；
5//! - 事件不丢：有存储时先落盘再投递，重启后从已确认游标续投；没有存储时进有界内存队列，满了背压
6//!   （见 [`queue`]）；
7//! - 配置了密钥时带 `X-Firecrawl-Signature: sha256=<HMAC-SHA256(密钥, 请求体)>`，和 Firecrawl 一样；
8//! - 目标地址过 SSRF 校验：请求时校验 URL，投递时的 DNS 解析结果逐个校验，连接的正是校验过的地址，
9//!   不跟随重定向，不走环境变量里的代理。
10
11use std::collections::BTreeMap;
12use std::sync::Arc;
13use std::time::Duration;
14
15use pagesift_net::{DnsLookup, SsrfPolicy, SystemLookup};
16use serde::{Deserialize, Serialize};
17use serde_json::{Map, Value};
18
19use crate::error::{ApiError, ErrorCode};
20use crate::metrics::Metrics;
21
22mod deliver;
23mod queue;
24#[cfg(test)]
25mod tests;
26
27pub(crate) use queue::{Hook, ack_key, outbox_key, seq_key};
28
29/// 请求里的 `webhook`：Firecrawl 接受一个 URL 字符串，或者带 headers、metadata、events 的对象。
30#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
31pub struct WebhookConfig {
32    pub url: String,
33    #[serde(default)]
34    pub headers: BTreeMap<String, String>,
35    #[serde(default)]
36    pub metadata: Map<String, Value>,
37    #[serde(default = "all_events")]
38    pub events: Vec<String>,
39}
40
41const EVENTS: [&str; 4] = ["started", "page", "completed", "failed"];
42
43fn all_events() -> Vec<String> {
44    EVENTS.iter().map(|e| e.to_string()).collect()
45}
46
47impl WebhookConfig {
48    /// 解析并校验请求里的 `webhook` 字段（含 SSRF 校验）。未知字段忽略。
49    pub(crate) fn parse(v: &Value, policy: &SsrfPolicy) -> Result<Self, ApiError> {
50        let cfg: Self = match v {
51            Value::String(u) => Self {
52                url: u.clone(),
53                headers: BTreeMap::new(),
54                metadata: Map::new(),
55                events: all_events(),
56            },
57            Value::Object(_) => serde_json::from_value(v.clone())
58                .map_err(|e| ApiError::invalid(format!("invalid webhook: {e}")))?,
59            _ => {
60                return Err(ApiError::invalid(
61                    "webhook must be a URL or an object with a url",
62                ));
63            }
64        };
65        if let Some(e) = cfg.events.iter().find(|e| !EVENTS.contains(&e.as_str())) {
66            return Err(ApiError::invalid(format!(
67                "unknown webhook event {e:?}: use started, page, completed or failed"
68            )));
69        }
70        for (k, v) in &cfg.headers {
71            if reqwest::header::HeaderName::from_bytes(k.as_bytes()).is_err()
72                || reqwest::header::HeaderValue::from_str(v).is_err()
73            {
74                return Err(ApiError::invalid(format!("invalid webhook header {k:?}")));
75            }
76        }
77        policy.check_url(&cfg.url).map_err(|e| {
78            ApiError::new(
79                ErrorCode::SsrfBlocked,
80                format!("webhook URL is not allowed: {e}"),
81            )
82        })?;
83        Ok(cfg)
84    }
85
86    /// 监控的 `webhook`：事件是 `monitor.page`、`monitor.check.completed`（照 Firecrawl），默认两个都要。
87    pub(crate) fn parse_monitor(v: &Value, policy: &SsrfPolicy) -> Result<Self, ApiError> {
88        const MONITOR_EVENTS: [&str; 2] = ["monitor.page", "monitor.check.completed"];
89        let mut obj = match v {
90            Value::String(u) => serde_json::json!({ "url": u }),
91            Value::Object(_) => v.clone(),
92            _ => {
93                return Err(ApiError::invalid(
94                    "webhook must be a URL or an object with a url",
95                ));
96            }
97        };
98        let events = obj.get("events").cloned();
99        obj["events"] = serde_json::json!(["started"]);
100        let mut cfg = Self::parse(&obj, policy)?;
101        cfg.events = match events {
102            Some(e) => serde_json::from_value(e)
103                .map_err(|e| ApiError::invalid(format!("invalid webhook events: {e}")))?,
104            None => MONITOR_EVENTS.iter().map(|e| e.to_string()).collect(),
105        };
106        if let Some(e) = cfg
107            .events
108            .iter()
109            .find(|e| !MONITOR_EVENTS.contains(&e.as_str()))
110        {
111            return Err(ApiError::invalid(format!(
112                "unknown monitor webhook event {e:?}: use monitor.page or monitor.check.completed"
113            )));
114        }
115        Ok(cfg)
116    }
117
118    /// agent 的 `webhook`：事件是 `started`、`action`、`completed`、`failed`、`cancelled`（也接受
119    /// `agent.started` 这样的全名，发送时总用全名），默认都要。
120    pub(crate) fn parse_agent(v: &Value, policy: &SsrfPolicy) -> Result<Self, ApiError> {
121        const AGENT_EVENTS: [&str; 5] = ["started", "action", "completed", "failed", "cancelled"]; // [协议] Firecrawl 的五个事件
122        let mut obj = match v {
123            Value::String(u) => serde_json::json!({ "url": u }),
124            Value::Object(_) => v.clone(),
125            _ => {
126                return Err(ApiError::invalid(
127                    "webhook must be a URL or an object with a url",
128                ));
129            }
130        };
131        let events = obj.get("events").cloned();
132        obj["events"] = serde_json::json!(["started"]);
133        let mut cfg = Self::parse(&obj, policy)?;
134        cfg.events = match events {
135            Some(e) => serde_json::from_value::<Vec<String>>(e)
136                .map_err(|e| ApiError::invalid(format!("invalid webhook events: {e}")))?
137                .into_iter()
138                .map(|e| e.strip_prefix("agent.").map(str::to_string).unwrap_or(e))
139                .collect(),
140            None => AGENT_EVENTS.iter().map(|e| e.to_string()).collect(),
141        };
142        if let Some(e) = cfg
143            .events
144            .iter()
145            .find(|e| !AGENT_EVENTS.contains(&e.as_str()))
146        {
147            return Err(ApiError::invalid(format!(
148                "unknown agent webhook event {e:?}: use started, action, completed, failed or cancelled"
149            )));
150        }
151        Ok(cfg)
152    }
153
154    fn wants(&self, event: Event) -> bool {
155        self.events.iter().any(|e| e == event.name())
156    }
157}
158
159#[derive(Debug, Clone, Copy, PartialEq, Eq)]
160pub(crate) enum Event {
161    Started,
162    Page,
163    Completed,
164    Failed,
165}
166
167impl Event {
168    fn name(self) -> &'static str {
169        match self {
170            Event::Started => "started",
171            Event::Page => "page",
172            Event::Completed => "completed",
173            Event::Failed => "failed",
174        }
175    }
176}
177
178/// webhook 投递的设置（服务端配置，不来自请求）。
179#[derive(Clone)]
180pub struct WebhookSettings {
181    /// 目标地址的 SSRF 策略，默认拒绝一切非公网地址。
182    pub ssrf: SsrfPolicy,
183    pub dns: Arc<dyn DnsLookup>,
184    /// 每个事件最多投递几次（含第一次）。
185    pub max_attempts: u32,
186    /// 第一次重试前等多久，之后每次翻倍，不超过 `max_backoff`。
187    pub initial_backoff: Duration,
188    pub max_backoff: Duration,
189    /// 单次投递的超时。
190    pub timeout: Duration,
191    /// 没有存储时每个任务排队等待投递的事件数上限，满了发送方等待（背压），不丢事件。
192    /// 有存储时事件排在盘上，不受这个限制。
193    pub queue: usize,
194    /// 签名密钥：设了就给每个请求加 `X-Firecrawl-Signature`，接收方用它验证来源。
195    pub secret: Option<String>,
196}
197
198impl Default for WebhookSettings {
199    fn default() -> Self {
200        Self {
201            ssrf: SsrfPolicy::default(),
202            dns: Arc::new(SystemLookup),
203            // [策略] 下面都是 `[server.webhook]` 可配的运维策略，不做在线估计：接收端是运维方自己的服务，
204            // 每个任务只有几次投递样本，生存估计收敛不了；事件已经落盘不丢，这几项只影响送达快慢。
205            // 默认值：指数退避 1 s 起、每次翻倍、封顶 60 s，5 次覆盖约 1+2+4+8+16 = 31 s 的短暂故障
206            // （AWS Architecture Blog, "Exponential Backoff and Jitter", 2015 的做法）；
207            // 单次超时 10 s 与 Firecrawl 的 webhook 超时一致；内存队列 1024 只在存储出错时使用，
208            // 按每个事件约 1 KiB 计，单个任务最多占 1 MiB。
209            max_attempts: 5,
210            initial_backoff: Duration::from_secs(1),
211            max_backoff: Duration::from_secs(60),
212            timeout: Duration::from_secs(10),
213            queue: 1024,
214            secret: None,
215        }
216    }
217}
218
219impl std::fmt::Debug for WebhookSettings {
220    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
221        f.debug_struct("WebhookSettings")
222            .field("ssrf", &self.ssrf)
223            .field("max_attempts", &self.max_attempts)
224            .field("initial_backoff", &self.initial_backoff)
225            .field("max_backoff", &self.max_backoff)
226            .field("timeout", &self.timeout)
227            .field("queue", &self.queue)
228            .field("secret", &self.secret.is_some())
229            .finish_non_exhaustive()
230    }
231}
232
233/// 所有任务共用的投递器（HTTP 客户端和设置）。
234#[derive(Clone)]
235pub(crate) struct Webhooks {
236    client: reqwest::Client,
237    pub(crate) settings: Arc<WebhookSettings>,
238    metrics: Arc<Metrics>,
239    /// 多服务器网关：落盘队列的写入同时复制到温备（与任务记录同一个镜像）。
240    pub(crate) mirror: Option<Arc<crate::jobs::mirror::JobMirror>>,
241}
242
243impl Webhooks {
244    pub(crate) fn new(settings: WebhookSettings, metrics: Arc<Metrics>) -> Self {
245        Self {
246            client: deliver::client(&settings),
247            settings: Arc::new(settings),
248            metrics,
249            mirror: None,
250        }
251    }
252
253    pub(crate) fn ssrf(&self) -> &SsrfPolicy {
254        &self.settings.ssrf
255    }
256}
pagesift_api/webhook/mod.rs

pagesift_api/webhook/
mod.rs
