pagesift_api/monitor/
mod.rs

1//! 监控（Firecrawl 的 `/v2/monitor` 形状）：按计划重复检查网址，发现变化时通知。
2//!
3//! - `POST /v2/monitor` 创建，`GET /v2/monitor` 列出，`GET/DELETE /v2/monitor/{id}`，
4//!   `GET /v2/monitor/{id}/checks`、`GET /v2/monitor/{id}/checks/{checkId}` 看检查结果；
5//!   `/v2/monitors` 是同样的别名。
6//! - 目标：`{"type": "scrape", "urls": [...], "scrapeOptions": {...}}`，或
7//!   `{"type": "crawl", "url": "...", "crawlOptions": {...}}`（每次检查爬一遍，比较新增、变化、消失的页面）。
8//!   `search` 目标不支持（如实报错）。
9//! - 计划：给了 `schedule.cron` 或 `schedule.text` 就按那个周期（和 Firecrawl 一样）；pagesift 扩展
10//!   `schedule.minIntervalMs`、`schedule.maxIntervalMs`（毫秒）或不给计划时，每个网址的间隔按变化率的后验
11//!   自适应（`pagesift_stat::poisson`），下界 5 分钟（Firecrawl 的最短间隔）。
12//! - 检查带 `If-None-Match` / `If-Modified-Since`；变化用块级、区分噪声的 diff 判断，默认只有实质变化
13//!   才算 `changed`（pagesift 扩展 `notify: "any"` 时噪声变化也算）；配了 `extract` 格式时附字段级变化。
14//! - 通知：`webhook`（`monitor.page`、`monitor.check.completed`，载荷照 Firecrawl）。邮件、Slack、
15//!   `goal` 的 LLM 判断不支持：`notification` 报错，`goal` 照存但在 `warnings` 里说明没有判断。
16//! - 服务端调度；多服务器时只有持有网关租约的机器调度（`Config::monitor_leader`）。
17
18mod crawl;
19mod run;
20pub mod schedule;
21
22use std::collections::{BTreeMap, HashMap, VecDeque};
23use std::sync::Arc;
24
25use axum::Json;
26use axum::extract::rejection::JsonRejection;
27use axum::extract::{Extension, Path, State};
28use serde::{Deserialize, Serialize};
29use serde_json::{Value, json};
30use tokio::sync::{Mutex, Notify};
31
32use crate::error::{ApiError, ErrorCode};
33use crate::limits::{Caller, KeyLimits};
34use crate::server::{AppState, lenient_body};
35use crate::types::ScrapeOptions;
36use crate::webhook::WebhookConfig;
37
38pub(crate) use run::start_scheduler;
39
40/// 监控能做什么判断：多服务器时由网关租约决定本机是否调度。
41pub type Leader = Arc<dyn Fn() -> bool + Send + Sync>;
42
43/// 请求体（Firecrawl 字段加 pagesift 扩展）。
44#[derive(Debug, Clone, Serialize, Deserialize)]
45#[serde(rename_all = "camelCase")]
46pub struct MonitorRequest {
47    #[serde(default)]
48    pub name: Option<String>,
49    #[serde(default)]
50    pub schedule: Option<Schedule>,
51    #[serde(default)]
52    pub goal: Option<String>,
53    #[serde(default)]
54    pub judge_enabled: Option<bool>,
55    pub targets: Vec<Target>,
56    #[serde(default)]
57    pub webhook: Option<Value>,
58    #[serde(default)]
59    pub notification: Option<Value>,
60    #[serde(default)]
61    pub retention_days: Option<u32>,
62    /// pagesift 扩展：`substantive`（默认，噪声变化不算）或 `any`。
63    #[serde(default)]
64    pub notify: Option<String>,
65}
66
67#[derive(Debug, Clone, Default, Serialize, Deserialize)]
68#[serde(rename_all = "camelCase")]
69pub struct Schedule {
70    #[serde(default, skip_serializing_if = "Option::is_none")]
71    pub cron: Option<String>,
72    #[serde(default, skip_serializing_if = "Option::is_none")]
73    pub text: Option<String>,
74    #[serde(default, skip_serializing_if = "Option::is_none")]
75    pub timezone: Option<String>,
76    /// pagesift 扩展：最短、最长检查间隔（毫秒）。
77    #[serde(default, skip_serializing_if = "Option::is_none")]
78    pub min_interval_ms: Option<u64>,
79    #[serde(default, skip_serializing_if = "Option::is_none")]
80    pub max_interval_ms: Option<u64>,
81}
82
83#[derive(Debug, Clone, Serialize, Deserialize)]
84#[serde(rename_all = "camelCase")]
85pub struct Target {
86    #[serde(default)]
87    pub id: Option<String>,
88    #[serde(rename = "type")]
89    pub kind: String,
90    #[serde(default)]
91    pub urls: Vec<String>,
92    #[serde(default, skip_serializing_if = "Option::is_none")]
93    pub url: Option<String>,
94    #[serde(default, skip_serializing_if = "Option::is_none")]
95    pub scrape_options: Option<ScrapeOptions>,
96    #[serde(default, skip_serializing_if = "Option::is_none")]
97    pub crawl_options: Option<Value>,
98}
99
100/// 一个网址的记录：上次的内容、条件请求用的校验值、变化率的观测、下次到期时间。
101#[derive(Debug, Clone, Default, Serialize, Deserialize)]
102#[serde(rename_all = "camelCase")]
103pub(crate) struct PageState {
104    pub markdown: Option<String>,
105    pub extract: Option<Value>,
106    pub etag: Option<String>,
107    pub last_modified: Option<String>,
108    /// 上次拿到结果的检查（Unix 秒）：变化率的观测区间从这里算，抓取失败不动它。
109    pub last_check: Option<f64>,
110    /// 上次尝试（Unix 秒，失败也算）：下次到期时间从这里算。
111    #[serde(default)]
112    pub last_attempt: Option<f64>,
113    /// 已经报过 `removed`：还是 404 就不再报。
114    #[serde(default)]
115    pub removed: bool,
116    /// 变化率观测（`pagesift_stat::poisson::ChangeRate` 的字段）：没变的总时长、各个变了的区间。
117    pub exposure: Option<f64>,
118    #[serde(default)]
119    pub changed_intervals: Vec<f64>,
120    /// 缓存：变化率后验均值（每秒）和 δ 规则的间隔（秒），重新分配预算时用。
121    #[serde(default)]
122    pub rate_mean: Option<f64>,
123    #[serde(default)]
124    pub delta_interval: Option<f64>,
125    /// 爬取目标里的页面跟着目标检查，是 INFINITY；JSON 里写成 null，读回时还原。
126    #[serde(deserialize_with = "due_or_never")]
127    pub next_due: f64,
128    /// crawl 目标：这个网址来自哪个目标（消失时报 removed）。
129    #[serde(default)]
130    pub target: String,
131}
132
133fn due_or_never<'de, D: serde::Deserializer<'de>>(d: D) -> Result<f64, D::Error> {
134    Ok(Option::<f64>::deserialize(d)?.unwrap_or(f64::INFINITY))
135}
136
137#[derive(Debug, Clone, Serialize, Deserialize)]
138#[serde(rename_all = "camelCase")]
139pub(crate) struct PageResult {
140    pub url: String,
141    /// `same`、`new`、`changed`、`removed`、`error`。
142    pub status: String,
143    #[serde(skip_serializing_if = "Option::is_none")]
144    pub error: Option<String>,
145    /// 只有噪声变化（日期、计数等），按 `notify` 不算 changed。
146    #[serde(skip_serializing_if = "std::ops::Not::not", default)]
147    pub noise_only: bool,
148    #[serde(skip_serializing_if = "Option::is_none")]
149    pub diff: Option<Value>,
150    #[serde(skip_serializing_if = "Vec::is_empty", default)]
151    pub field_changes: Vec<Value>,
152    /// 用的是条件请求的 304。
153    #[serde(skip_serializing_if = "std::ops::Not::not", default)]
154    pub not_modified: bool,
155}
156
157#[derive(Debug, Clone, Serialize, Deserialize)]
158#[serde(rename_all = "camelCase")]
159pub(crate) struct Check {
160    pub id: String,
161    pub started_at: String,
162    pub finished_at: String,
163    pub status: String,
164    pub pages: Vec<PageResult>,
165}
166
167impl Check {
168    fn summary(&self) -> Value {
169        let n = |s: &str| self.pages.iter().filter(|p| p.status == s).count();
170        json!({
171            "totalPages": self.pages.len(), "same": n("same"), "changed": n("changed"),
172            "new": n("new"), "removed": n("removed"), "error": n("error"),
173        })
174    }
175}
176
177#[derive(Debug, Clone, Serialize, Deserialize)]
178#[serde(rename_all = "camelCase")]
179pub(crate) struct MonitorState {
180    pub id: String,
181    pub owner: String,
182    pub req: MonitorRequest,
183    pub created_at: String,
184    pub status: String,
185    /// 固定周期（秒）：给了 cron 或文本时；自适应时为 `None`。
186    pub period: Option<u64>,
187    pub min_interval: u64,
188    pub max_interval: Option<u64>,
189    pub webhook: Option<WebhookConfig>,
190    pub pages: BTreeMap<String, PageState>,
191    pub checks: VecDeque<Check>,
192    pub last_run_at: Option<String>,
193    pub warnings: Vec<String>,
194}
195
196impl MonitorState {
197    pub(crate) fn next_due(&self) -> Option<f64> {
198        self.pages
199            .values()
200            .map(|p| p.next_due)
201            .fold(None, |a, b| Some(a.map_or(b, |a: f64| a.min(b))))
202    }
203
204    fn view(&self) -> Value {
205        let schedule = match self.period {
206            Some(p) => {
207                // cron 原样回显；时刻和时区不生效时 `warnings` 里有说明，实际按 UTC 的周期算
208                let given = self.req.schedule.as_ref().and_then(|s| s.cron.clone());
209                json!({ "cron": given.unwrap_or_else(|| schedule::normalized_cron(p)), "timezone": "UTC" })
210            }
211            None => {
212                json!({ "adaptive": true, "minIntervalMs": self.min_interval * 1000, "maxIntervalMs": self.max_interval.map(|m| m * 1000) })
213            }
214        };
215        let next = self.next_due().map(run::rfc3339_secs);
216        let mut v = json!({
217            "id": self.id, "name": self.req.name, "status": self.status, "schedule": schedule,
218            "nextRunAt": next, "lastRunAt": self.last_run_at, "goal": self.req.goal,
219            "judgeEnabled": false, "targets": self.req.targets, "webhook": self.webhook,
220            "retentionDays": self.req.retention_days.unwrap_or(DEFAULT_RETENTION_DAYS),
221            "createdAt": self.created_at,
222        });
223        if !self.warnings.is_empty() {
224            v["warnings"] = json!(self.warnings);
225        }
226        v
227    }
228}
229
230/// [协议] Firecrawl 文档：`retentionDays` 默认 30，最多 365。
231const DEFAULT_RETENTION_DAYS: u32 = 30;
232const MAX_RETENTION_DAYS: u32 = 365; // [协议] 同上
233/// [协议] Firecrawl 文档：每个监控 1–50 个目标。
234const MAX_TARGETS: usize = 50;
235/// [策略] 一个账户最多这么多个监控：每个监控都存页面全文，常驻内存
236const MAX_MONITORS_PER_ACCOUNT: usize = 1000;
237/// [策略] 每个监控保留的检查记录条数上限，和 `retentionDays` 同时生效：5 分钟一次时一年有 10 万次
238const MAX_CHECKS_KEPT: usize = 1000;
239
240/// 所有监控（进程内），配了存储时同时落盘（`monitor/<id>`）。
241#[derive(Clone, Default)]
242pub(crate) struct Monitors {
243    pub map: Arc<Mutex<HashMap<String, MonitorState>>>,
244    pub wake: Arc<Notify>,
245    /// 调度任务在跑：只在有监控时运行（它持有服务状态和存储，空闲时不该占着）。
246    pub running: Arc<std::sync::atomic::AtomicBool>,
247    pub leader: Arc<std::sync::Mutex<Option<Leader>>>,
248    /// 各账户的限额：后台检查按监控的主人扣页面配额。
249    pub accounts: Arc<HashMap<String, Arc<KeyLimits>>>,
250    pub default_limits: Arc<KeyLimits>,
251}
252
253impl Monitors {
254    fn caller(&self, owner: &str) -> Caller {
255        let limits = self.accounts.get(owner).unwrap_or(&self.default_limits);
256        Caller::named(owner, limits.clone())
257    }
258}
259
260const KNOWN: &[&str] = &[
261    "name",
262    "schedule",
263    "goal",
264    "judgeEnabled",
265    "targets",
266    "webhook",
267    "notification",
268    "retentionDays",
269    "notify",
270];
271
272fn owner_of(caller: &Option<Extension<Caller>>) -> String {
273    caller.as_ref().map_or_else(
274        || "anonymous".to_string(),
275        |Extension(c)| c.account.to_string(),
276    )
277}
278
279pub(crate) async fn create(
280    State(s): State<AppState>,
281    caller: Option<Extension<Caller>>,
282    req: Result<Json<Value>, JsonRejection>,
283) -> Result<Json<Value>, ApiError> {
284    let (req, ignored): (MonitorRequest, _) =
285        lenient_body(req, &[KNOWN, crate::fields::CLIENT_IDENTITY])?;
286    let mut warnings: Vec<String> = crate::fields::ignored_warning(&ignored)
287        .into_iter()
288        .collect();
289    if req.targets.is_empty() || req.targets.len() > MAX_TARGETS {
290        return Err(ApiError::invalid(format!(
291            "a monitor needs 1 to {MAX_TARGETS} targets"
292        )));
293    }
294    if req.notification.is_some() {
295        return Err(ApiError::unsupported(
296            "notification (email, Slack) is not supported: use a webhook",
297        ));
298    }
299    if req.judge_enabled == Some(true) {
300        return Err(ApiError::unsupported(
301            "judgeEnabled: LLM judging of changes is not supported; changes are classified by a noise-aware block diff",
302        ));
303    }
304    if req.goal.is_some() {
305        warnings.push(
306            "monitor_goal_not_judged: the goal is stored but changes are not judged against it"
307                .into(),
308        );
309    }
310    let notify = req.notify.as_deref().unwrap_or("substantive");
311    if !matches!(notify, "substantive" | "any") {
312        return Err(ApiError::invalid("notify must be substantive or any"));
313    }
314    if req
315        .retention_days
316        .is_some_and(|d| d == 0 || d > MAX_RETENTION_DAYS)
317    {
318        return Err(ApiError::invalid(format!(
319            "retentionDays must be between 1 and {MAX_RETENTION_DAYS}"
320        )));
321    }
322    let sched = req.schedule.clone().unwrap_or_default();
323    let period = match (&sched.cron, &sched.text) {
324        (Some(c), _) => Some(schedule::cron_period(c)?),
325        (None, Some(t)) => Some(schedule::text_period(t)?),
326        (None, None) => None,
327    };
328    if period.is_some_and(|p| p < schedule::MIN_INTERVAL_SECS) {
329        return Err(ApiError::invalid("the minimum interval is 5 minutes"));
330    }
331    if period.is_some_and(|p| p > schedule::MAX_INTERVAL_SECS)
332        || sched
333            .min_interval_ms
334            .into_iter()
335            .chain(sched.max_interval_ms)
336            .any(|ms| ms / 1000 > schedule::MAX_INTERVAL_SECS)
337    {
338        return Err(ApiError::invalid(format!(
339            "the maximum interval is {} days",
340            schedule::MAX_INTERVAL_SECS / 86_400
341        )));
342    }
343    warnings.extend(schedule::unsupported(&sched, period));
344    let adaptive =
345        sched.min_interval_ms.is_some() || sched.max_interval_ms.is_some() || period.is_none();
346    let min_interval = sched
347        .min_interval_ms
348        .map_or(schedule::MIN_INTERVAL_SECS, |ms| ms.div_ceil(1000))
349        .max(schedule::MIN_INTERVAL_SECS);
350    let max_interval = sched.max_interval_ms.map(|ms| ms / 1000).or(period);
351    if max_interval.is_some_and(|m| m < min_interval) {
352        return Err(ApiError::invalid(
353            "maxIntervalMs must not be shorter than minIntervalMs",
354        ));
355    }
356    let webhook = match &req.webhook {
357        Some(v) => Some(WebhookConfig::parse_monitor(v, &s.webhooks.settings.ssrf)?),
358        None => None,
359    };
360    crate::validate::check_url_count(req.targets.iter().map(|t| t.urls.len()).sum())?;
361    let owner = owner_of(&caller);
362    let held = s
363        .monitors
364        .map
365        .lock()
366        .await
367        .values()
368        .filter(|m| m.owner == owner)
369        .count();
370    if held >= MAX_MONITORS_PER_ACCOUNT {
371        return Err(ApiError::new(
372            ErrorCode::ConcurrencyLimitExceeded,
373            format!("this account already has {held} monitors, the maximum"),
374        ));
375    }
376    let now = run::now_secs();
377    let id = uuid::Uuid::new_v4().to_string();
378    let mut pages = BTreeMap::new();
379    let mut targets = req.targets.clone();
380    for t in &mut targets {
381        t.id.get_or_insert_with(|| uuid::Uuid::new_v4().to_string());
382        let opts = t.scrape_options.clone().unwrap_or_default();
383        crate::validate::check_scrape_options(&opts, s.engine.as_ref())?;
384        match t.kind.as_str() {
385            "scrape" => {
386                if t.urls.is_empty() {
387                    return Err(ApiError::invalid("a scrape target needs urls"));
388                }
389                for u in &t.urls {
390                    crate::validate::check_url(u)?;
391                    pages.insert(
392                        u.clone(),
393                        PageState {
394                            next_due: now,
395                            target: t.id.clone().unwrap_or_default(),
396                            ..Default::default()
397                        },
398                    );
399                }
400            }
401            "crawl" => {
402                let u = t
403                    .url
404                    .clone()
405                    .ok_or_else(|| ApiError::invalid("a crawl target needs url"))?;
406                crate::validate::check_url(&u)?;
407                crawl::crawl_request(t, &u).and_then(|r| {
408                    crate::validate::check_url_count(r.limit)?;
409                    crate::firecrawl::crawl::check_crawl_request(&r, s.engine.as_ref())
410                })?;
411                // 爬取目标的“网址”是起始页：整个目标一起检查，变化率按“有没有任何页面变化”记
412                pages.insert(
413                    format!("crawl:{}", t.id.clone().unwrap_or_default()),
414                    PageState {
415                        next_due: now,
416                        target: t.id.clone().unwrap_or_default(),
417                        ..Default::default()
418                    },
419                );
420            }
421            "search" => {
422                return Err(ApiError::unsupported(
423                    "search targets are not supported; monitor known URLs (scrape) or a site (crawl)",
424                ));
425            }
426            other => return Err(ApiError::invalid(format!("unknown target type {other:?}"))),
427        }
428    }
429    let mut req = req;
430    req.targets = targets;
431    let state = MonitorState {
432        id: id.clone(),
433        owner,
434        req,
435        created_at: run::rfc3339_secs(now),
436        status: "active".into(),
437        period: (!adaptive).then_some(period).flatten(),
438        min_interval,
439        max_interval,
440        webhook,
441        pages,
442        checks: VecDeque::new(),
443        last_run_at: None,
444        warnings,
445    };
446    let view = state.view();
447    run::persist(&s, &state);
448    s.monitors.map.lock().await.insert(id, state);
449    s.monitors.wake.notify_one();
450    run::ensure_running(&s);
451    Ok(Json(json!({ "success": true, "data": view })))
452}
453
454async fn owned<'a>(
455    map: &'a mut HashMap<String, MonitorState>,
456    id: &str,
457    owner: &str,
458) -> Result<&'a mut MonitorState, ApiError> {
459    map.get_mut(id)
460        .filter(|m| m.owner == owner)
461        .ok_or_else(|| ApiError::new(ErrorCode::NotFound, format!("monitor {id} not found")))
462}
463
464pub(crate) async fn list(
465    State(s): State<AppState>,
466    caller: Option<Extension<Caller>>,
467) -> Json<Value> {
468    let owner = owner_of(&caller);
469    let map = s.monitors.map.lock().await;
470    let mut all: Vec<&MonitorState> = map.values().filter(|m| m.owner == owner).collect();
471    all.sort_by(|a, b| a.created_at.cmp(&b.created_at));
472    Json(json!({ "success": true, "data": all.iter().map(|m| m.view()).collect::<Vec<_>>() }))
473}
474
475pub(crate) async fn get(
476    State(s): State<AppState>,
477    caller: Option<Extension<Caller>>,
478    Path(id): Path<String>,
479) -> Result<Json<Value>, ApiError> {
480    let mut map = s.monitors.map.lock().await;
481    let m = owned(&mut map, &id, &owner_of(&caller)).await?;
482    Ok(Json(json!({ "success": true, "data": m.view() })))
483}
484
485pub(crate) async fn delete(
486    State(s): State<AppState>,
487    caller: Option<Extension<Caller>>,
488    Path(id): Path<String>,
489) -> Result<Json<Value>, ApiError> {
490    let mut map = s.monitors.map.lock().await;
491    owned(&mut map, &id, &owner_of(&caller)).await?;
492    map.remove(&id);
493    run::forget(&s, &id);
494    // 调度任务在等下一个到期时间：叫醒它，监控清空时让它退出
495    s.monitors.wake.notify_one();
496    Ok(Json(json!({ "success": true })))
497}
498
499pub(crate) async fn checks(
500    State(s): State<AppState>,
501    caller: Option<Extension<Caller>>,
502    Path(id): Path<String>,
503) -> Result<Json<Value>, ApiError> {
504    let mut map = s.monitors.map.lock().await;
505    let m = owned(&mut map, &id, &owner_of(&caller)).await?;
506    let data: Vec<Value> = m
507        .checks
508        .iter()
509        .rev()
510        .map(|c| json!({ "id": c.id, "monitorId": m.id, "status": c.status, "startedAt": c.started_at, "finishedAt": c.finished_at, "summary": c.summary() }))
511        .collect();
512    Ok(Json(json!({ "success": true, "data": data })))
513}
514
515pub(crate) async fn check(
516    State(s): State<AppState>,
517    caller: Option<Extension<Caller>>,
518    Path((id, check_id)): Path<(String, String)>,
519) -> Result<Json<Value>, ApiError> {
520    let mut map = s.monitors.map.lock().await;
521    let m = owned(&mut map, &id, &owner_of(&caller)).await?;
522    let c =
523        m.checks.iter().find(|c| c.id == check_id).ok_or_else(|| {
524            ApiError::new(ErrorCode::NotFound, format!("check {check_id} not found"))
525        })?;
526    Ok(Json(json!({ "success": true, "data": {
527        "id": c.id, "monitorId": m.id, "status": c.status, "startedAt": c.started_at,
528        "finishedAt": c.finished_at, "summary": c.summary(), "pages": c.pages,
529    }})))
530}
531
532/// pagesift 扩展 `POST /v2/monitor/{id}/run`：现在就检查全部网址（不等计划），返回这次的检查结果。
533pub(crate) async fn run_now(
534    State(s): State<AppState>,
535    caller: Option<Extension<Caller>>,
536    Path(id): Path<String>,
537) -> Result<Json<Value>, ApiError> {
538    {
539        let mut map = s.monitors.map.lock().await;
540        let m = owned(&mut map, &id, &owner_of(&caller)).await?;
541        let now = run::now_secs();
542        for p in m.pages.values_mut().filter(|p| p.next_due.is_finite()) {
543            p.next_due = p.next_due.min(now);
544        }
545    }
546    run::run_check(&s, &id).await;
547    let map = s.monitors.map.lock().await;
548    let m = map
549        .get(&id)
550        .ok_or_else(|| ApiError::new(ErrorCode::NotFound, format!("monitor {id} not found")))?;
551    let c = m.checks.back().cloned();
552    Ok(Json(json!({ "success": true, "data": c.map(|c| json!({
553        "id": c.id, "status": c.status, "summary": c.summary(), "pages": c.pages,
554        "nextRunAt": m.next_due().map(run::rfc3339_secs),
555    })) })))
556}
557
558pub(crate) fn routes() -> axum::Router<AppState> {
559    use axum::routing::get as g;
560    let mut r = axum::Router::new();
561    for base in ["/v2/monitor", "/v2/monitors"] {
562        r = r
563            .route(base, axum::routing::post(create).get(list))
564            .route(&format!("{base}/{{id}}"), g(get).delete(delete))
565            .route(&format!("{base}/{{id}}/checks"), g(checks))
566            .route(&format!("{base}/{{id}}/run"), axum::routing::post(run_now))
567            .route(&format!("{base}/{{id}}/checks/{{check_id}}"), g(check));
568    }
569    r
570}
571
572#[cfg(test)]
573mod tests;
pagesift_api/monitor/mod.rs

pagesift_api/monitor/
mod.rs
