mod crawl;
mod run;
pub mod schedule;
use std::collections::{BTreeMap, HashMap, VecDeque};
use std::sync::Arc;
use axum::Json;
use axum::extract::rejection::JsonRejection;
use axum::extract::{Extension, Path, State};
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use tokio::sync::{Mutex, Notify};
use crate::error::{ApiError, ErrorCode};
use crate::limits::{Caller, KeyLimits};
use crate::server::{AppState, lenient_body};
use crate::types::ScrapeOptions;
use crate::webhook::WebhookConfig;
pub(crate) use run::start_scheduler;
pub type Leader = Arc<dyn Fn() -> bool + Send + Sync>;
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct MonitorRequest {
#[serde(default)]
pub name: Option<String>,
#[serde(default)]
pub schedule: Option<Schedule>,
#[serde(default)]
pub goal: Option<String>,
#[serde(default)]
pub judge_enabled: Option<bool>,
pub targets: Vec<Target>,
#[serde(default)]
pub webhook: Option<Value>,
#[serde(default)]
pub notification: Option<Value>,
#[serde(default)]
pub retention_days: Option<u32>,
#[serde(default)]
pub notify: Option<String>,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct Schedule {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub cron: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub text: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub timezone: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub min_interval_ms: Option<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub max_interval_ms: Option<u64>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct Target {
#[serde(default)]
pub id: Option<String>,
#[serde(rename = "type")]
pub kind: String,
#[serde(default)]
pub urls: Vec<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub url: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub scrape_options: Option<ScrapeOptions>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub crawl_options: Option<Value>,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct PageState {
pub markdown: Option<String>,
pub extract: Option<Value>,
pub etag: Option<String>,
pub last_modified: Option<String>,
pub last_check: Option<f64>,
#[serde(default)]
pub last_attempt: Option<f64>,
#[serde(default)]
pub removed: bool,
pub exposure: Option<f64>,
#[serde(default)]
pub changed_intervals: Vec<f64>,
#[serde(default)]
pub rate_mean: Option<f64>,
#[serde(default)]
pub delta_interval: Option<f64>,
#[serde(deserialize_with = "due_or_never")]
pub next_due: f64,
#[serde(default)]
pub target: String,
}
fn due_or_never<'de, D: serde::Deserializer<'de>>(d: D) -> Result<f64, D::Error> {
Ok(Option::<f64>::deserialize(d)?.unwrap_or(f64::INFINITY))
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct PageResult {
pub url: String,
pub status: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub error: Option<String>,
#[serde(skip_serializing_if = "std::ops::Not::not", default)]
pub noise_only: bool,
#[serde(skip_serializing_if = "Option::is_none")]
pub diff: Option<Value>,
#[serde(skip_serializing_if = "Vec::is_empty", default)]
pub field_changes: Vec<Value>,
#[serde(skip_serializing_if = "std::ops::Not::not", default)]
pub not_modified: bool,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct Check {
pub id: String,
pub started_at: String,
pub finished_at: String,
pub status: String,
pub pages: Vec<PageResult>,
}
impl Check {
fn summary(&self) -> Value {
let n = |s: &str| self.pages.iter().filter(|p| p.status == s).count();
json!({
"totalPages": self.pages.len(), "same": n("same"), "changed": n("changed"),
"new": n("new"), "removed": n("removed"), "error": n("error"),
})
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct MonitorState {
pub id: String,
pub owner: String,
pub req: MonitorRequest,
pub created_at: String,
pub status: String,
pub period: Option<u64>,
pub min_interval: u64,
pub max_interval: Option<u64>,
pub webhook: Option<WebhookConfig>,
pub pages: BTreeMap<String, PageState>,
pub checks: VecDeque<Check>,
pub last_run_at: Option<String>,
pub warnings: Vec<String>,
}
impl MonitorState {
pub(crate) fn next_due(&self) -> Option<f64> {
self.pages
.values()
.map(|p| p.next_due)
.fold(None, |a, b| Some(a.map_or(b, |a: f64| a.min(b))))
}
fn view(&self) -> Value {
let schedule = match self.period {
Some(p) => {
let given = self.req.schedule.as_ref().and_then(|s| s.cron.clone());
json!({ "cron": given.unwrap_or_else(|| schedule::normalized_cron(p)), "timezone": "UTC" })
}
None => {
json!({ "adaptive": true, "minIntervalMs": self.min_interval * 1000, "maxIntervalMs": self.max_interval.map(|m| m * 1000) })
}
};
let next = self.next_due().map(run::rfc3339_secs);
let mut v = json!({
"id": self.id, "name": self.req.name, "status": self.status, "schedule": schedule,
"nextRunAt": next, "lastRunAt": self.last_run_at, "goal": self.req.goal,
"judgeEnabled": false, "targets": self.req.targets, "webhook": self.webhook,
"retentionDays": self.req.retention_days.unwrap_or(DEFAULT_RETENTION_DAYS),
"createdAt": self.created_at,
});
if !self.warnings.is_empty() {
v["warnings"] = json!(self.warnings);
}
v
}
}
const DEFAULT_RETENTION_DAYS: u32 = 30;
const MAX_RETENTION_DAYS: u32 = 365; const MAX_TARGETS: usize = 50;
const MAX_MONITORS_PER_ACCOUNT: usize = 1000;
const MAX_CHECKS_KEPT: usize = 1000;
#[derive(Clone, Default)]
pub(crate) struct Monitors {
pub map: Arc<Mutex<HashMap<String, MonitorState>>>,
pub wake: Arc<Notify>,
pub running: Arc<std::sync::atomic::AtomicBool>,
pub leader: Arc<std::sync::Mutex<Option<Leader>>>,
pub accounts: Arc<HashMap<String, Arc<KeyLimits>>>,
pub default_limits: Arc<KeyLimits>,
}
impl Monitors {
fn caller(&self, owner: &str) -> Caller {
let limits = self.accounts.get(owner).unwrap_or(&self.default_limits);
Caller::named(owner, limits.clone())
}
}
const KNOWN: &[&str] = &[
"name",
"schedule",
"goal",
"judgeEnabled",
"targets",
"webhook",
"notification",
"retentionDays",
"notify",
];
fn owner_of(caller: &Option<Extension<Caller>>) -> String {
caller.as_ref().map_or_else(
|| "anonymous".to_string(),
|Extension(c)| c.account.to_string(),
)
}
pub(crate) async fn create(
State(s): State<AppState>,
caller: Option<Extension<Caller>>,
req: Result<Json<Value>, JsonRejection>,
) -> Result<Json<Value>, ApiError> {
let (req, ignored): (MonitorRequest, _) =
lenient_body(req, &[KNOWN, crate::fields::CLIENT_IDENTITY])?;
let mut warnings: Vec<String> = crate::fields::ignored_warning(&ignored)
.into_iter()
.collect();
if req.targets.is_empty() || req.targets.len() > MAX_TARGETS {
return Err(ApiError::invalid(format!(
"a monitor needs 1 to {MAX_TARGETS} targets"
)));
}
if req.notification.is_some() {
return Err(ApiError::unsupported(
"notification (email, Slack) is not supported: use a webhook",
));
}
if req.judge_enabled == Some(true) {
return Err(ApiError::unsupported(
"judgeEnabled: LLM judging of changes is not supported; changes are classified by a noise-aware block diff",
));
}
if req.goal.is_some() {
warnings.push(
"monitor_goal_not_judged: the goal is stored but changes are not judged against it"
.into(),
);
}
let notify = req.notify.as_deref().unwrap_or("substantive");
if !matches!(notify, "substantive" | "any") {
return Err(ApiError::invalid("notify must be substantive or any"));
}
if req
.retention_days
.is_some_and(|d| d == 0 || d > MAX_RETENTION_DAYS)
{
return Err(ApiError::invalid(format!(
"retentionDays must be between 1 and {MAX_RETENTION_DAYS}"
)));
}
let sched = req.schedule.clone().unwrap_or_default();
let period = match (&sched.cron, &sched.text) {
(Some(c), _) => Some(schedule::cron_period(c)?),
(None, Some(t)) => Some(schedule::text_period(t)?),
(None, None) => None,
};
if period.is_some_and(|p| p < schedule::MIN_INTERVAL_SECS) {
return Err(ApiError::invalid("the minimum interval is 5 minutes"));
}
if period.is_some_and(|p| p > schedule::MAX_INTERVAL_SECS)
|| sched
.min_interval_ms
.into_iter()
.chain(sched.max_interval_ms)
.any(|ms| ms / 1000 > schedule::MAX_INTERVAL_SECS)
{
return Err(ApiError::invalid(format!(
"the maximum interval is {} days",
schedule::MAX_INTERVAL_SECS / 86_400
)));
}
warnings.extend(schedule::unsupported(&sched, period));
let adaptive =
sched.min_interval_ms.is_some() || sched.max_interval_ms.is_some() || period.is_none();
let min_interval = sched
.min_interval_ms
.map_or(schedule::MIN_INTERVAL_SECS, |ms| ms.div_ceil(1000))
.max(schedule::MIN_INTERVAL_SECS);
let max_interval = sched.max_interval_ms.map(|ms| ms / 1000).or(period);
if max_interval.is_some_and(|m| m < min_interval) {
return Err(ApiError::invalid(
"maxIntervalMs must not be shorter than minIntervalMs",
));
}
let webhook = match &req.webhook {
Some(v) => Some(WebhookConfig::parse_monitor(v, &s.webhooks.settings.ssrf)?),
None => None,
};
crate::validate::check_url_count(req.targets.iter().map(|t| t.urls.len()).sum())?;
let owner = owner_of(&caller);
let held = s
.monitors
.map
.lock()
.await
.values()
.filter(|m| m.owner == owner)
.count();
if held >= MAX_MONITORS_PER_ACCOUNT {
return Err(ApiError::new(
ErrorCode::ConcurrencyLimitExceeded,
format!("this account already has {held} monitors, the maximum"),
));
}
let now = run::now_secs();
let id = uuid::Uuid::new_v4().to_string();
let mut pages = BTreeMap::new();
let mut targets = req.targets.clone();
for t in &mut targets {
t.id.get_or_insert_with(|| uuid::Uuid::new_v4().to_string());
let opts = t.scrape_options.clone().unwrap_or_default();
crate::validate::check_scrape_options(&opts, s.engine.as_ref())?;
match t.kind.as_str() {
"scrape" => {
if t.urls.is_empty() {
return Err(ApiError::invalid("a scrape target needs urls"));
}
for u in &t.urls {
crate::validate::check_url(u)?;
pages.insert(
u.clone(),
PageState {
next_due: now,
target: t.id.clone().unwrap_or_default(),
..Default::default()
},
);
}
}
"crawl" => {
let u = t
.url
.clone()
.ok_or_else(|| ApiError::invalid("a crawl target needs url"))?;
crate::validate::check_url(&u)?;
crawl::crawl_request(t, &u).and_then(|r| {
crate::validate::check_url_count(r.limit)?;
crate::firecrawl::crawl::check_crawl_request(&r, s.engine.as_ref())
})?;
pages.insert(
format!("crawl:{}", t.id.clone().unwrap_or_default()),
PageState {
next_due: now,
target: t.id.clone().unwrap_or_default(),
..Default::default()
},
);
}
"search" => {
return Err(ApiError::unsupported(
"search targets are not supported; monitor known URLs (scrape) or a site (crawl)",
));
}
other => return Err(ApiError::invalid(format!("unknown target type {other:?}"))),
}
}
let mut req = req;
req.targets = targets;
let state = MonitorState {
id: id.clone(),
owner,
req,
created_at: run::rfc3339_secs(now),
status: "active".into(),
period: (!adaptive).then_some(period).flatten(),
min_interval,
max_interval,
webhook,
pages,
checks: VecDeque::new(),
last_run_at: None,
warnings,
};
let view = state.view();
run::persist(&s, &state);
s.monitors.map.lock().await.insert(id, state);
s.monitors.wake.notify_one();
run::ensure_running(&s);
Ok(Json(json!({ "success": true, "data": view })))
}
async fn owned<'a>(
map: &'a mut HashMap<String, MonitorState>,
id: &str,
owner: &str,
) -> Result<&'a mut MonitorState, ApiError> {
map.get_mut(id)
.filter(|m| m.owner == owner)
.ok_or_else(|| ApiError::new(ErrorCode::NotFound, format!("monitor {id} not found")))
}
pub(crate) async fn list(
State(s): State<AppState>,
caller: Option<Extension<Caller>>,
) -> Json<Value> {
let owner = owner_of(&caller);
let map = s.monitors.map.lock().await;
let mut all: Vec<&MonitorState> = map.values().filter(|m| m.owner == owner).collect();
all.sort_by(|a, b| a.created_at.cmp(&b.created_at));
Json(json!({ "success": true, "data": all.iter().map(|m| m.view()).collect::<Vec<_>>() }))
}
pub(crate) async fn get(
State(s): State<AppState>,
caller: Option<Extension<Caller>>,
Path(id): Path<String>,
) -> Result<Json<Value>, ApiError> {
let mut map = s.monitors.map.lock().await;
let m = owned(&mut map, &id, &owner_of(&caller)).await?;
Ok(Json(json!({ "success": true, "data": m.view() })))
}
pub(crate) async fn delete(
State(s): State<AppState>,
caller: Option<Extension<Caller>>,
Path(id): Path<String>,
) -> Result<Json<Value>, ApiError> {
let mut map = s.monitors.map.lock().await;
owned(&mut map, &id, &owner_of(&caller)).await?;
map.remove(&id);
run::forget(&s, &id);
s.monitors.wake.notify_one();
Ok(Json(json!({ "success": true })))
}
pub(crate) async fn checks(
State(s): State<AppState>,
caller: Option<Extension<Caller>>,
Path(id): Path<String>,
) -> Result<Json<Value>, ApiError> {
let mut map = s.monitors.map.lock().await;
let m = owned(&mut map, &id, &owner_of(&caller)).await?;
let data: Vec<Value> = m
.checks
.iter()
.rev()
.map(|c| json!({ "id": c.id, "monitorId": m.id, "status": c.status, "startedAt": c.started_at, "finishedAt": c.finished_at, "summary": c.summary() }))
.collect();
Ok(Json(json!({ "success": true, "data": data })))
}
pub(crate) async fn check(
State(s): State<AppState>,
caller: Option<Extension<Caller>>,
Path((id, check_id)): Path<(String, String)>,
) -> Result<Json<Value>, ApiError> {
let mut map = s.monitors.map.lock().await;
let m = owned(&mut map, &id, &owner_of(&caller)).await?;
let c =
m.checks.iter().find(|c| c.id == check_id).ok_or_else(|| {
ApiError::new(ErrorCode::NotFound, format!("check {check_id} not found"))
})?;
Ok(Json(json!({ "success": true, "data": {
"id": c.id, "monitorId": m.id, "status": c.status, "startedAt": c.started_at,
"finishedAt": c.finished_at, "summary": c.summary(), "pages": c.pages,
}})))
}
pub(crate) async fn run_now(
State(s): State<AppState>,
caller: Option<Extension<Caller>>,
Path(id): Path<String>,
) -> Result<Json<Value>, ApiError> {
{
let mut map = s.monitors.map.lock().await;
let m = owned(&mut map, &id, &owner_of(&caller)).await?;
let now = run::now_secs();
for p in m.pages.values_mut().filter(|p| p.next_due.is_finite()) {
p.next_due = p.next_due.min(now);
}
}
run::run_check(&s, &id).await;
let map = s.monitors.map.lock().await;
let m = map
.get(&id)
.ok_or_else(|| ApiError::new(ErrorCode::NotFound, format!("monitor {id} not found")))?;
let c = m.checks.back().cloned();
Ok(Json(json!({ "success": true, "data": c.map(|c| json!({
"id": c.id, "status": c.status, "summary": c.summary(), "pages": c.pages,
"nextRunAt": m.next_due().map(run::rfc3339_secs),
})) })))
}
pub(crate) fn routes() -> axum::Router<AppState> {
use axum::routing::get as g;
let mut r = axum::Router::new();
for base in ["/v2/monitor", "/v2/monitors"] {
r = r
.route(base, axum::routing::post(create).get(list))
.route(&format!("{base}/{{id}}"), g(get).delete(delete))
.route(&format!("{base}/{{id}}/checks"), g(checks))
.route(&format!("{base}/{{id}}/run"), axum::routing::post(run_now))
.route(&format!("{base}/{{id}}/checks/{{check_id}}"), g(check));
}
r
}
#[cfg(test)]
mod tests;