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    fn wants(&self, event: Event) -> bool {
87        self.events.iter().any(|e| e == event.name())
88    }
89}
90
91#[derive(Debug, Clone, Copy, PartialEq, Eq)]
92pub(crate) enum Event {
93    Started,
94    Page,
95    Completed,
96    Failed,
97}
98
99impl Event {
100    fn name(self) -> &'static str {
101        match self {
102            Event::Started => "started",
103            Event::Page => "page",
104            Event::Completed => "completed",
105            Event::Failed => "failed",
106        }
107    }
108}
109
110/// webhook 投递的设置（服务端配置，不来自请求）。
111#[derive(Clone)]
112pub struct WebhookSettings {
113    /// 目标地址的 SSRF 策略，默认拒绝一切非公网地址。
114    pub ssrf: SsrfPolicy,
115    pub dns: Arc<dyn DnsLookup>,
116    /// 每个事件最多投递几次（含第一次）。
117    pub max_attempts: u32,
118    /// 第一次重试前等多久，之后每次翻倍，不超过 `max_backoff`。
119    pub initial_backoff: Duration,
120    pub max_backoff: Duration,
121    /// 单次投递的超时。
122    pub timeout: Duration,
123    /// 没有存储时每个任务排队等待投递的事件数上限，满了发送方等待（背压），不丢事件。
124    /// 有存储时事件排在盘上，不受这个限制。
125    pub queue: usize,
126    /// 签名密钥：设了就给每个请求加 `X-Firecrawl-Signature`，接收方用它验证来源。
127    pub secret: Option<String>,
128}
129
130impl Default for WebhookSettings {
131    fn default() -> Self {
132        Self {
133            ssrf: SsrfPolicy::default(),
134            dns: Arc::new(SystemLookup),
135            // [策略] 下面都是 `[server.webhook]` 可配的运维策略，不做在线估计：接收端是运维方自己的服务，
136            // 每个任务只有几次投递样本，生存估计收敛不了；事件已经落盘不丢，这几项只影响送达快慢。
137            // 默认值：指数退避 1 s 起、每次翻倍、封顶 60 s，5 次覆盖约 1+2+4+8+16 = 31 s 的短暂故障
138            // （AWS Architecture Blog, "Exponential Backoff and Jitter", 2015 的做法）；
139            // 单次超时 10 s 与 Firecrawl 的 webhook 超时一致；内存队列 1024 只在存储出错时使用，
140            // 按每个事件约 1 KiB 计，单个任务最多占 1 MiB。
141            max_attempts: 5,
142            initial_backoff: Duration::from_secs(1),
143            max_backoff: Duration::from_secs(60),
144            timeout: Duration::from_secs(10),
145            queue: 1024,
146            secret: None,
147        }
148    }
149}
150
151impl std::fmt::Debug for WebhookSettings {
152    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
153        f.debug_struct("WebhookSettings")
154            .field("ssrf", &self.ssrf)
155            .field("max_attempts", &self.max_attempts)
156            .field("initial_backoff", &self.initial_backoff)
157            .field("max_backoff", &self.max_backoff)
158            .field("timeout", &self.timeout)
159            .field("queue", &self.queue)
160            .field("secret", &self.secret.is_some())
161            .finish_non_exhaustive()
162    }
163}
164
165/// 所有任务共用的投递器（HTTP 客户端和设置）。
166#[derive(Clone)]
167pub(crate) struct Webhooks {
168    client: reqwest::Client,
169    settings: Arc<WebhookSettings>,
170    metrics: Arc<Metrics>,
171    /// 多服务器网关：落盘队列的写入同时复制到温备（与任务记录同一个镜像）。
172    pub(crate) mirror: Option<Arc<crate::jobs::mirror::JobMirror>>,
173}
174
175impl Webhooks {
176    pub(crate) fn new(settings: WebhookSettings, metrics: Arc<Metrics>) -> Self {
177        Self {
178            client: deliver::client(&settings),
179            settings: Arc::new(settings),
180            metrics,
181            mirror: None,
182        }
183    }
184
185    pub(crate) fn ssrf(&self) -> &SsrfPolicy {
186        &self.settings.ssrf
187    }
188}
pagesift_api/webhook/mod.rs

pagesift_api/webhook/
mod.rs
