AnyBot/src/triggers.rs

883 lines
45 KiB
Rust
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

// Event-triggered routines (the original's trigger kinds, main-app.cjs routine trigger schema:
// cron, slack, github, origin, microsoftTeams, linear, sentry, pagerduty, email, webhook, group).
// The original's servers receive those events; here AnyBot checks for them with the person's own
// tokens while it runs (the original also keeps event triggers running locally when it can't
// sync: 「事件觸發的例行任務在安全的情況下會繼續在本機執行」), and a webhook address is served on
// this Mac. When one fires, the routine's bot is woken with what happened.
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use std::collections::{HashMap, HashSet};
use std::io::{Read, Write};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::mpsc::Sender;
use std::sync::{Arc, Mutex, OnceLock};
use std::time::Duration;
/// GitHub asks clients to poll events at most every 60 s (X-Poll-Interval).
pub const GITHUB_POLL_SECS: u64 = 60;
/// Linear, Sentry, PagerDuty, Mail (estimated).
pub const SERVICE_POLL_SECS: u64 = 60;
/// Calendar checks (estimated).
pub const CALENDAR_POLL_SECS: u64 = 30;
/// Slack channel checks (estimated).
pub const SLACK_POLL_SECS: u64 = 10;
/// The webhook server on this Mac (estimated; next to the egress proxy's 8791).
pub const WEBHOOK_PORT: u16 = 8792;
pub const GITHUB_EVENTS: [&str; 15] = [
"pr_opened", "pr_updated", "pr_merged", "pr_closed", "pr_comment", "ci_passed", "ci_failed", "review_approved", "review_changes_requested", "review_comment", "thread_resolved", "thread_reopened",
"issue_assigned", "issue_opened", "push",
];
#[derive(Clone, Serialize, Deserialize, Debug, PartialEq)]
#[serde(tag = "type", rename_all = "lowercase")]
pub enum Trigger {
Github {
repo: String,
#[serde(default)]
token: String,
#[serde(default)]
events: Vec<String>,
/// Only these people's actions (empty = anyone).
#[serde(default)]
users: Vec<String>,
/// Only this branch (CI / pushes); empty = any.
#[serde(default)]
branch: String,
/// Only this PR number (0 = any).
#[serde(default)]
pr: u64,
},
Slack {
channel: String,
/// Empty = the token of a Slack 訊息管道.
#[serde(default)]
token: String,
/// any | mention | keyword | reaction
#[serde(default = "any", rename = "match")]
matcher: String,
/// The keyword or the reaction name.
#[serde(default)]
value: String,
},
Linear {
#[serde(rename = "apiKey")]
api_key: String,
/// issue_created | status_changed
#[serde(default)]
events: Vec<String>,
#[serde(default)]
team: String,
},
Sentry {
token: String,
org: String,
#[serde(default)]
project: String,
/// issue_created | issue_resolved | issue_unresolved | issue_ignored | issue_assigned | issue_changed
#[serde(default)]
events: Vec<String>,
},
Pagerduty {
token: String,
#[serde(default)]
service: String,
/// triggered | acknowledged | resolved | changed
#[serde(default)]
events: Vec<String>,
},
Email {
/// Sender contains (empty = anyone).
#[serde(default)]
from: String,
#[serde(default)]
subject: String,
},
Webhook {
/// Callers send `Authorization: Bearer <secret>`.
secret: String,
},
/// 跟著行事曆: when a tracked Calendar event starts, wherever it has been moved to.
Calendar {
/// Only this calendar (empty = all).
#[serde(default)]
calendar: String,
/// Only events whose title contains this (empty = all).
#[serde(default)]
title: String,
/// Fire this many minutes before the start (0 = at the start).
#[serde(default, rename = "minutesBefore")]
minutes_before: i64,
},
Group {
triggers: Vec<Trigger>,
},
}
fn any() -> String {
"any".into()
}
impl Trigger {
pub fn describe(&self) -> String {
match self {
Trigger::Github { repo, events, .. } => format!("GitHub {repo}:{}", if events.is_empty() { "任何事件".into() } else { events.join("、") }),
Trigger::Slack { channel, matcher, value, .. } => format!(
"Slack {channel}:{}",
match matcher.as_str() {
"mention" => "被 @ 時".to_string(),
"keyword" => format!("提到「{value}」時"),
"reaction" => format!("有人加上 :{value}: 時"),
_ => "任何訊息".to_string(),
}
),
Trigger::Linear { team, events, .. } => format!("Linear{}:{}", if team.is_empty() { String::new() } else { format!(" {team}") }, if events.is_empty() { "議題建立或狀態變更".into() } else { events.join("、") }),
Trigger::Sentry { org, project, events, .. } => format!("Sentry {org}/{}:{}", if project.is_empty() { "*" } else { project }, if events.is_empty() { "任何議題變化".into() } else { events.join("、") }),
Trigger::Pagerduty { service, events, .. } => format!("PagerDuty{}:{}", if service.is_empty() { String::new() } else { format!(" {service}") }, if events.is_empty() { "任何事故變化".into() } else { events.join("、") }),
Trigger::Email { from, subject } => format!("新郵件{}{}", if from.is_empty() { String::new() } else { format!(",寄件人含「{from}」") }, if subject.is_empty() { String::new() } else { format!(",主旨含「{subject}」") }),
Trigger::Webhook { .. } => "Webhook 呼叫".into(),
Trigger::Calendar { calendar, title, minutes_before } => format!(
"行事曆{}{}活動{}",
if calendar.is_empty() { String::new() } else { format!("「{calendar}」") },
if title.is_empty() { String::new() } else { format!("中標題含「{title}」的") },
if *minutes_before > 0 { format!("開始前 {minutes_before} 分鐘") } else { "開始時".into() }
),
Trigger::Group { triggers } => format!("任一:{}", triggers.iter().map(|t| t.describe()).collect::<Vec<_>>().join(";")),
}
}
/// Check what can be checked before saving.
pub fn validate(&self) -> Result<(), String> {
match self {
Trigger::Github { repo, events, .. } => {
if repo.split('/').filter(|p| !p.is_empty()).count() != 2 {
return Err("repo 要寫成 owner/name".into());
}
if let Some(e) = events.iter().find(|e| !GITHUB_EVENTS.contains(&e.as_str())) {
return Err(format!("不認得 GitHub 事件「{e}」,可以用:{}", GITHUB_EVENTS.join("、")));
}
}
Trigger::Slack { channel, matcher, .. } => {
if channel.trim().is_empty() {
return Err("Slack 要給頻道 ID".into());
}
if !["any", "mention", "keyword", "reaction"].contains(&matcher.as_str()) {
return Err("match 要是 any、mention、keyword 或 reaction".into());
}
}
Trigger::Linear { api_key, .. } if api_key.trim().is_empty() => return Err("Linear 要給 API key".into()),
Trigger::Sentry { token, org, .. } if token.trim().is_empty() || org.trim().is_empty() => return Err("Sentry 要給 token 和 org".into()),
Trigger::Pagerduty { token, .. } if token.trim().is_empty() => return Err("PagerDuty 要給 token".into()),
Trigger::Webhook { secret } if secret.trim().len() < 8 => return Err("Webhook 的 secret 至少 8 個字".into()),
Trigger::Group { triggers } => {
if triggers.len() < 2 {
return Err("群組至少要兩個觸發條件".into());
}
for t in triggers {
t.validate()?;
}
}
_ => {}
}
Ok(())
}
}
/// The JSON schema the bot fills in (create_routine / update_routine).
pub fn schema() -> Value {
json!({
"type": "object",
"description": "Run when something happens instead of on a schedule. type: github | slack | linear | sentry | pagerduty | email | webhook | group.",
"properties": {
"type": {"type": "string", "enum": ["github", "slack", "linear", "sentry", "pagerduty", "email", "webhook", "calendar", "group"]},
"repo": {"type": "string", "description": "github: owner/name"},
"token": {"type": "string", "description": "github/sentry/pagerduty token; slack: optional (uses a Slack 訊息管道's token)"},
"apiKey": {"type": "string", "description": "linear API key"},
"events": {"type": "array", "items": {"type": "string"}, "description": format!("github: {}; linear: issue_created, status_changed; sentry: issue_created, issue_resolved, issue_unresolved, issue_ignored, issue_assigned, issue_changed; pagerduty: triggered, acknowledged, resolved, changed. Empty = all.", GITHUB_EVENTS.join(", "))},
"users": {"type": "array", "items": {"type": "string"}, "description": "github: only these logins"},
"branch": {"type": "string"}, "pr": {"type": "integer"},
"channel": {"type": "string", "description": "slack channel id"},
"match": {"type": "string", "enum": ["any", "mention", "keyword", "reaction"]},
"value": {"type": "string", "description": "slack keyword or reaction name"},
"team": {"type": "string"}, "org": {"type": "string"}, "project": {"type": "string"}, "service": {"type": "string"},
"from": {"type": "string", "description": "email: sender contains"}, "subject": {"type": "string", "description": "email: subject contains"},
"secret": {"type": "string", "description": "webhook: bearer secret (at least 8 characters)"},
"calendar": {"type": "string", "description": "calendar: only this calendar"}, "title": {"type": "string", "description": "calendar: event title contains"}, "minutesBefore": {"type": "integer", "description": "calendar: fire this many minutes before the event starts"},
"triggers": {"type": "array", "items": {"type": "object"}, "description": "group: two or more triggers"}
},
"required": ["type"]
})
}
// ---------- reading events (pure, testable) ----------
/// One thing that happened: (event kind, who did it, branch, PR/issue number, description).
#[derive(Clone, Debug, PartialEq)]
pub struct Event {
pub kind: String,
pub user: String,
pub branch: String,
pub number: u64,
pub text: String,
}
/// GitHub repo events newer than `since` (event ids grow), oldest first.
pub fn github_events(list: &Value, since: u64) -> (u64, Vec<Event>) {
let mut newest = since;
let mut out = vec![];
let Some(arr) = list.as_array() else { return (since, out) };
let mut items: Vec<&Value> = arr.iter().collect();
items.sort_by_key(|e| e["id"].as_str().and_then(|s| s.parse::<u64>().ok()).unwrap_or(0));
for e in items {
let id = e["id"].as_str().and_then(|s| s.parse::<u64>().ok()).unwrap_or(0);
if id <= since {
continue;
}
newest = newest.max(id);
let user = e["actor"]["login"].as_str().unwrap_or("").to_string();
let p = &e["payload"];
let action = p["action"].as_str().unwrap_or("");
let pr = &p["pull_request"];
let (kind, number, text, branch) = match e["type"].as_str().unwrap_or("") {
"PullRequestEvent" => {
let k = match action {
"opened" | "reopened" => "pr_opened",
"closed" if pr["merged"].as_bool() == Some(true) => "pr_merged",
"closed" => "pr_closed",
"synchronize" | "edited" | "ready_for_review" => "pr_updated",
_ => continue,
};
(k, pr["number"].as_u64().unwrap_or(0), format!("PR #{} {}", pr["number"], pr["title"].as_str().unwrap_or("")), pr["head"]["ref"].as_str().unwrap_or("").to_string())
}
"IssueCommentEvent" if p["issue"]["pull_request"].is_object() => ("pr_comment", p["issue"]["number"].as_u64().unwrap_or(0), format!("PR #{} 的留言:{}", p["issue"]["number"], p["comment"]["body"].as_str().unwrap_or("").chars().take(200).collect::<String>()), String::new()),
"PullRequestReviewEvent" => {
let k = match p["review"]["state"].as_str().unwrap_or("").to_lowercase().as_str() {
"approved" => "review_approved",
"changes_requested" => "review_changes_requested",
"commented" => "review_comment",
_ => continue,
};
(k, pr["number"].as_u64().unwrap_or(0), format!("PR #{} 的審查:{}", pr["number"], p["review"]["body"].as_str().unwrap_or("").chars().take(200).collect::<String>()), pr["head"]["ref"].as_str().unwrap_or("").to_string())
}
"PullRequestReviewThreadEvent" => {
let k = if action == "resolved" { "thread_resolved" } else if action == "unresolved" { "thread_reopened" } else { continue };
(k, pr["number"].as_u64().unwrap_or(0), format!("PR #{} 的審查討論串{}", pr["number"], if k == "thread_resolved" { "已解決" } else { "重新開啟" }), String::new())
}
"IssuesEvent" => {
let k = match action {
"assigned" => "issue_assigned",
"opened" => "issue_opened",
_ => continue,
};
(k, p["issue"]["number"].as_u64().unwrap_or(0), format!("議題 #{} {}{}", p["issue"]["number"], p["issue"]["title"].as_str().unwrap_or(""), if k == "issue_assigned" { format!("(指派給 {})", p["assignee"]["login"].as_str().unwrap_or("")) } else { String::new() }), String::new())
}
"PushEvent" => {
let branch = p["ref"].as_str().unwrap_or("").trim_start_matches("refs/heads/").to_string();
("push", 0, format!("{} 推送到 {branch}:{} 個 commit", user, p["size"].as_u64().or(p["commits"].as_array().map(|a| a.len() as u64)).unwrap_or(0)), branch)
}
_ => continue,
};
out.push(Event { kind: kind.into(), user, branch, number, text });
}
(newest, out)
}
/// Finished GitHub Actions runs not seen before → ci_passed / ci_failed.
pub fn github_runs(runs: &Value, seen: &mut HashSet<u64>, first: bool) -> Vec<Event> {
let mut out = vec![];
for r in runs["workflow_runs"].as_array().into_iter().flatten() {
if r["status"].as_str() != Some("completed") {
continue;
}
let id = r["id"].as_u64().unwrap_or(0);
if !seen.insert(id) || first {
continue;
}
let kind = match r["conclusion"].as_str().unwrap_or("") {
"success" => "ci_passed",
"failure" | "timed_out" => "ci_failed",
_ => continue,
};
let pr = r["pull_requests"][0]["number"].as_u64().unwrap_or(0);
out.push(Event {
kind: kind.into(),
user: r["actor"]["login"].as_str().unwrap_or("").into(),
branch: r["head_branch"].as_str().unwrap_or("").into(),
number: pr,
text: format!("CI「{}」在 {} {}:{}", r["name"].as_str().unwrap_or(""), r["head_branch"].as_str().unwrap_or(""), if kind == "ci_passed" { "通過" } else { "失敗" }, r["html_url"].as_str().unwrap_or("")),
});
}
out
}
/// A list of (id, status, title) compared with the last one: new ids (after the first look) and
/// status changes. Events are `created` and `<status>`.
pub fn status_changes(prev: &mut HashMap<String, String>, items: &[(String, String, String)], first: bool) -> Vec<(String, String)> {
let mut out = vec![];
// An empty prev is always the baseline: a poll that follows a failed first round must
// record, not fire every existing item as "created" (history replay).
let baseline = first || prev.is_empty();
for (id, status, title) in items {
match prev.get(id) {
None if !baseline => out.push(("created".to_string(), title.clone())),
Some(old) if old != status => out.push((status.clone(), format!("{title}({old} → {status})"))),
_ => {}
}
prev.insert(id.clone(), status.clone());
}
out
}
/// Calendar events that start now (within the last poll) and haven't fired: (key, description).
pub fn calendar_due(events: &Value, now: i64, minutes_before: i64, window_secs: i64, title: &str, fired: &mut HashSet<String>) -> Vec<String> {
let mut out = vec![];
for e in events.as_array().into_iter().flatten() {
let name = e["title"].as_str().unwrap_or("");
if !title.is_empty() && !name.to_lowercase().contains(&title.to_lowercase()) {
continue;
}
let Some(start) = e["start"].as_str().and_then(|s| chrono::DateTime::parse_from_rfc3339(s).ok()).map(|d| d.timestamp()) else { continue };
let at = start - minutes_before * 60;
// The key includes the start: a moved event fires again at its new time.
let key = format!("{}@{start}", e["id"].as_str().unwrap_or(name));
if at <= now && now - at < window_secs && fired.insert(key) {
out.push(format!("行事曆活動「{name}」{}({})", if minutes_before > 0 { format!("{minutes_before} 分鐘後開始") } else { "開始了".into() }, e["start"].as_str().unwrap_or("")));
}
}
out
}
/// Does a Slack message match?
pub fn slack_matches(text: &str, matcher: &str, value: &str, own_user: &str) -> bool {
match matcher {
"mention" => !own_user.is_empty() && text.contains(&format!("<@{own_user}>")),
"keyword" => !value.is_empty() && text.to_lowercase().contains(&value.to_lowercase()),
"reaction" => false,
_ => true,
}
}
fn wanted(events: &[String], kind: &str) -> bool {
events.is_empty() || events.iter().any(|e| e == kind)
}
// ---------- running ----------
#[derive(Clone, Debug)]
pub struct Fired {
pub bot_id: u64,
pub routine_id: u64,
pub what: String,
}
#[derive(Clone)]
pub struct Watch {
pub bot_id: u64,
pub routine_id: u64,
pub trigger: Trigger,
}
static GENERATION: AtomicU64 = AtomicU64::new(0);
pub fn statuses() -> &'static Mutex<HashMap<u64, String>> {
static S: OnceLock<Mutex<HashMap<u64, String>>> = OnceLock::new();
S.get_or_init(|| Mutex::new(HashMap::new()))
}
fn note(routine: u64, s: &str) {
statuses().lock().unwrap().insert(routine, s.to_string());
}
fn webhooks() -> &'static Mutex<HashMap<u64, (String, u64)>> {
static W: OnceLock<Mutex<HashMap<u64, (String, u64)>>> = OnceLock::new();
W.get_or_init(|| Mutex::new(HashMap::new()))
}
fn agent(secs: u64) -> ureq::Agent {
ureq::Agent::config_builder().timeout_global(Some(Duration::from_secs(secs))).http_status_as_error(false).build().into()
}
fn get(url: &str, headers: &[(&str, String)]) -> Result<Value, String> {
let mut r = agent(30).get(url).header("User-Agent", "AnyBot");
for (k, v) in headers {
r = r.header(*k, v);
}
let mut resp = r.call().map_err(|e| e.to_string())?;
let st = resp.status().as_u16();
let body = resp.body_mut().read_to_string().unwrap_or_default();
if !(200..300).contains(&st) {
return Err(format!("HTTP {st}:{}", body.chars().take(150).collect::<String>()));
}
serde_json::from_str(&body).map_err(|e| e.to_string())
}
fn post(url: &str, headers: &[(&str, String)], body: &Value) -> Result<Value, String> {
let mut r = agent(30).post(url).header("User-Agent", "AnyBot").header("Content-Type", "application/json");
for (k, v) in headers {
r = r.header(*k, v);
}
let mut resp = r.send(body.to_string().as_bytes()).map_err(|e| e.to_string())?;
let st = resp.status().as_u16();
let text = resp.body_mut().read_to_string().unwrap_or_default();
if !(200..300).contains(&st) {
return Err(format!("HTTP {st}:{}", text.chars().take(150).collect::<String>()));
}
serde_json::from_str(&text).map_err(|e| e.to_string())
}
fn sleep_alive(gen: u64, secs: u64) -> bool {
for _ in 0..secs * 4 {
if GENERATION.load(Ordering::SeqCst) != gen {
return false;
}
std::thread::sleep(Duration::from_millis(250));
}
GENERATION.load(Ordering::SeqCst) == gen
}
/// (Re)start every watcher. `slack_token` is used when a Slack trigger has none.
pub fn restart(watches: Vec<Watch>, slack_token: String, tx: Sender<Fired>, repaint: Arc<dyn Fn() + Send + Sync>) {
let gen = GENERATION.fetch_add(1, Ordering::SeqCst) + 1;
webhooks().lock().unwrap().clear();
// Fresh router too: old entries held senders of the previous generation (a leak, and a
// webhook could fire a routine that no longer exists).
*ROUTER_TX.lock().unwrap() = None;
for w in watches {
start_one(w.bot_id, w.routine_id, w.trigger, gen, slack_token.clone(), tx.clone(), repaint.clone());
}
}
fn start_one(bot_id: u64, routine_id: u64, t: Trigger, gen: u64, slack_token: String, tx: Sender<Fired>, repaint: Arc<dyn Fn() + Send + Sync>) {
let fire = move |what: String| {
let _ = tx.send(Fired { bot_id, routine_id, what });
repaint();
};
match t {
Trigger::Group { triggers } => {
for t in triggers {
let tx2 = fire.clone();
let (stx, srx) = std::sync::mpsc::channel::<Fired>();
start_one(bot_id, routine_id, t, gen, slack_token.clone(), stx, Arc::new(|| {}));
std::thread::spawn(move || {
while let Ok(f) = srx.recv() {
tx2(f.what);
}
});
}
}
Trigger::Webhook { secret } => {
webhooks().lock().unwrap().insert(routine_id, (secret, bot_id));
match ensure_webhook_server(fire_router()) {
Ok(port) => note(routine_id, &format!("等待呼叫:POST http://127.0.0.1:{port}/hooks/{routine_id}")),
Err(e) => note(routine_id, &e),
}
// The router delivers through the shared sender set below.
ROUTER_TX.lock().unwrap().get_or_insert_with(HashMap::new).insert(routine_id, Box::new(fire));
}
other => {
std::thread::spawn(move || watch_loop(routine_id, other, gen, slack_token, fire));
}
}
}
type Fire = Box<dyn Fn(String) + Send>;
static ROUTER_TX: Mutex<Option<HashMap<u64, Fire>>> = Mutex::new(None);
fn fire_router() -> fn(u64, String) {
|id, what| {
if let Some(m) = ROUTER_TX.lock().unwrap().as_ref() {
if let Some(f) = m.get(&id) {
f(what);
}
}
}
}
fn watch_loop(routine_id: u64, t: Trigger, gen: u64, slack_token: String, fire: impl Fn(String)) {
let mut first = true;
let mut cursor: u64 = 0;
let mut seen: HashSet<u64> = HashSet::new();
let mut prev: HashMap<String, String> = HashMap::new();
let mut slack_cursor = String::new();
let mut slack_own = String::new();
let mut slack_reacted: HashSet<String> = HashSet::new();
loop {
// first must only advance after a round that actually reached the service: a failed
// first round used to make the next one replay the whole history as new events.
let mut reached = true;
let wait = match &t {
Trigger::Github { repo, token, events, users, branch, pr } => {
let mut h = vec![("Accept", "application/vnd.github+json".to_string())];
if !token.trim().is_empty() {
h.push(("Authorization", format!("Bearer {}", token.trim())));
}
let ok_ev = |e: &Event| wanted(events, &e.kind) && (users.is_empty() || users.iter().any(|u| u.eq_ignore_ascii_case(&e.user))) && (branch.is_empty() || e.branch.is_empty() || e.branch == *branch) && (*pr == 0 || e.number == *pr);
match get(&format!("https://api.github.com/repos/{repo}/events?per_page=50"), &h) {
Ok(v) => {
let (next, evs) = github_events(&v, cursor);
if !first {
for e in evs.into_iter().filter(ok_ev) {
fire(format!("GitHub {repo}:{}({},by {})", e.text, e.kind, e.user));
}
}
cursor = next;
note(routine_id, "已連上 GitHub,等待事件");
}
Err(e) => { reached = false; note(routine_id, &format!("GitHub:{e}")) },
}
if wanted(events, "ci_passed") || wanted(events, "ci_failed") {
if let Ok(v) = get(&format!("https://api.github.com/repos/{repo}/actions/runs?per_page=20"), &h) {
for e in github_runs(&v, &mut seen, first).into_iter().filter(ok_ev) {
fire(format!("GitHub {repo}:{}", e.text));
}
}
}
GITHUB_POLL_SECS
}
Trigger::Slack { channel, token, matcher, value } => {
let tok = if token.trim().is_empty() { slack_token.clone() } else { token.trim().to_string() };
let auth = vec![("Authorization", format!("Bearer {tok}"))];
if slack_own.is_empty() {
slack_own = get("https://slack.com/api/auth.test", &auth).ok().and_then(|v| v["user_id"].as_str().map(String::from)).unwrap_or_default();
}
match get(&format!("https://slack.com/api/conversations.history?channel={channel}&limit=30"), &auth) {
Ok(v) if v["ok"].as_bool() == Some(true) => {
let msgs: Vec<&Value> = v["messages"].as_array().map(|a| a.iter().rev().collect()).unwrap_or_default();
for m in &msgs {
let ts = m["ts"].as_str().unwrap_or("").to_string();
let text = m["text"].as_str().unwrap_or("");
if matcher == "reaction" {
for r in m["reactions"].as_array().into_iter().flatten() {
if r["name"].as_str() == Some(value.as_str()) {
let key = format!("{ts}:{}", r["count"]);
if slack_reacted.insert(key) && !first {
fire(format!("Slack {channel}:有人在訊息「{}」加上 :{value}:", text.chars().take(120).collect::<String>()));
}
}
}
continue;
}
let newer = ts.parse::<f64>().unwrap_or(0.0) > slack_cursor.parse::<f64>().unwrap_or(0.0);
if newer && !first && m.get("bot_id").is_none() && slack_matches(text, matcher, value, &slack_own) {
fire(format!("Slack {channel} 的 {}:{text}", m["user"].as_str().unwrap_or("someone")));
}
if newer {
slack_cursor = ts;
}
}
note(routine_id, "已連上 Slack,等待訊息");
}
Ok(v) => note(routine_id, &format!("Slack 說:{}", v["error"].as_str().unwrap_or("不明錯誤"))),
Err(e) => { reached = false; note(routine_id, &format!("Slack:{e}")) },
}
SLACK_POLL_SECS
}
Trigger::Linear { api_key, events, team } => {
let q = json!({"query": "{ issues(first: 50, orderBy: updatedAt) { nodes { id identifier title state { name } team { key } } } }"});
match post("https://api.linear.app/graphql", &[("Authorization", api_key.trim().to_string())], &q) {
Ok(v) => {
let items: Vec<(String, String, String)> = v["data"]["issues"]["nodes"]
.as_array()
.into_iter()
.flatten()
.filter(|n| team.is_empty() || n["team"]["key"].as_str() == Some(team.as_str()))
.map(|n| (n["id"].as_str().unwrap_or("").into(), n["state"]["name"].as_str().unwrap_or("").into(), format!("{} {}", n["identifier"].as_str().unwrap_or(""), n["title"].as_str().unwrap_or(""))))
.collect();
for (ev, title) in status_changes(&mut prev, &items, first) {
let kind = if ev == "created" { "issue_created" } else { "status_changed" };
if wanted(events, kind) {
fire(format!("Linear:{title}({})", if kind == "issue_created" { "新議題" } else { "狀態變更" }));
}
}
note(routine_id, "已連上 Linear,等待變化");
}
Err(e) => { reached = false; note(routine_id, &format!("Linear:{e}")) },
}
SERVICE_POLL_SECS
}
Trigger::Sentry { token, org, project, events } => {
let url = if project.is_empty() { format!("https://sentry.io/api/0/organizations/{org}/issues/?limit=50&query=") } else { format!("https://sentry.io/api/0/projects/{org}/{project}/issues/?limit=50&query=") };
match get(&url, &[("Authorization", format!("Bearer {}", token.trim()))]) {
Ok(v) => {
let items: Vec<(String, String, String)> = v
.as_array()
.into_iter()
.flatten()
.map(|i| {
let assigned = if i["assignedTo"].is_null() { "" } else { "+assigned" };
(i["id"].as_str().unwrap_or("").into(), format!("{}{assigned}", i["status"].as_str().unwrap_or("")), format!("{} {}", i["shortId"].as_str().unwrap_or(""), i["title"].as_str().unwrap_or("")))
})
.collect();
for (ev, title) in status_changes(&mut prev, &items, first) {
let kind = if ev == "created" {
"issue_created"
} else if ev.ends_with("+assigned") && !ev.starts_with("resolved") {
"issue_assigned"
} else if ev.starts_with("resolved") {
"issue_resolved"
} else if ev.starts_with("unresolved") {
"issue_unresolved"
} else if ev.starts_with("ignored") {
"issue_ignored"
} else {
"issue_changed"
};
if wanted(events, kind) || wanted(events, "issue_changed") && kind != "issue_created" {
fire(format!("Sentry:{title}({kind})"));
}
}
note(routine_id, "已連上 Sentry,等待變化");
}
Err(e) => { reached = false; note(routine_id, &format!("Sentry:{e}")) },
}
SERVICE_POLL_SECS
}
Trigger::Pagerduty { token, service, events } => {
let url = format!("https://api.pagerduty.com/incidents?limit=50&sort_by=created_at:desc{}", if service.is_empty() { String::new() } else { format!("&service_ids[]={service}") });
match get(&url, &[("Authorization", format!("Token token={}", token.trim())), ("Accept", "application/vnd.pagerduty+json;version=2".into())]) {
Ok(v) => {
let items: Vec<(String, String, String)> = v["incidents"]
.as_array()
.into_iter()
.flatten()
.map(|i| (i["id"].as_str().unwrap_or("").into(), i["status"].as_str().unwrap_or("").into(), format!("#{} {}", i["incident_number"], i["title"].as_str().unwrap_or(""))))
.collect();
for (ev, title) in status_changes(&mut prev, &items, first) {
let kind = if ev == "created" { "triggered".to_string() } else { ev.clone() };
if wanted(events, &kind) || wanted(events, "changed") && ev != "created" {
fire(format!("PagerDuty:{title}({kind})"));
}
}
note(routine_id, "已連上 PagerDuty,等待事故");
}
Err(e) => { reached = false; note(routine_id, &format!("PagerDuty:{e}")) },
}
SERVICE_POLL_SECS
}
Trigger::Email { from, subject } => {
let (o, ok) = crate::macapps::run("mac_read_mail", &json!({"limit": 20, "days": 2}).to_string());
if ok {
let list: Value = serde_json::from_str(o.trim_start_matches("郵件:")).unwrap_or(Value::Null);
let items: Vec<(String, String, String)> = list
.as_array()
.into_iter()
.flatten()
.filter(|m| from.is_empty() || m["from"].as_str().unwrap_or("").to_lowercase().contains(&from.to_lowercase()))
.filter(|m| subject.is_empty() || m["subject"].as_str().unwrap_or("").to_lowercase().contains(&subject.to_lowercase()))
.map(|m| (m["id"].as_str().unwrap_or("").into(), "in".into(), format!("{}:{}(id {})", m["from"].as_str().unwrap_or(""), m["subject"].as_str().unwrap_or(""), m["id"].as_str().unwrap_or(""))))
.collect();
for (_, title) in status_changes(&mut prev, &items, first) {
fire(format!("新郵件 {title}"));
}
note(routine_id, "正在看「郵件」的收件匣");
} else {
reached = false;
note(routine_id, &o);
}
SERVICE_POLL_SECS
}
Trigger::Calendar { calendar, title, minutes_before } => {
let now = chrono::Utc::now();
let from = (now - chrono::Duration::hours(1)).to_rfc3339();
let to = (now + chrono::Duration::hours(24)).to_rfc3339();
let (o, ok) = crate::macapps::run("mobile_read_calendar", &json!({"from": from, "to": to, "calendar": calendar}).to_string());
if ok {
let list: Value = serde_json::from_str(o.trim_start_matches("行事曆:")).unwrap_or(Value::Null);
for what in calendar_due(&list, now.timestamp(), *minutes_before, CALENDAR_POLL_SECS as i64 * 2, title, &mut slack_reacted) {
fire(what);
}
note(routine_id, "正在看「行事曆」");
} else {
reached = false;
note(routine_id, &o);
}
CALENDAR_POLL_SECS
}
Trigger::Webhook { .. } | Trigger::Group { .. } => return,
};
if reached {
first = false;
}
// These sets grow with every event seen: a watcher that runs for months must stay
// bounded (ids grow, so keep the newest range; the react/calendar sets are only
// short-term de-duplication, so a cap that clears is harmless).
if seen.len() > 4096 {
let m = *seen.iter().max().unwrap_or(&0);
seen.retain(|id| *id + 100_000 >= m);
}
if prev.len() > 4096 {
// status_changes treats an empty set as a quiet baseline (records, fires nothing),
// so clearing is safe and keeps the map from growing with every issue ever seen.
prev.clear();
}
if slack_reacted.len() > 4096 {
slack_reacted.clear();
}
if !sleep_alive(gen, wait) {
return;
}
}
}
// ---------- webhook server ----------
static WEBHOOK_BOUND: Mutex<Option<u16>> = Mutex::new(None);
pub fn webhook_port() -> Option<u16> {
*WEBHOOK_BOUND.lock().unwrap()
}
/// Parse `POST /hooks/<id>` with a bearer secret. Returns (routine id, secret, body).
pub fn parse_hook(req: &[u8]) -> Option<(u64, String, String)> {
let text = String::from_utf8_lossy(req);
let (head, body) = text.split_once("\r\n\r\n")?;
let first = head.lines().next()?;
let mut parts = first.split_whitespace();
if parts.next()? != "POST" {
return None;
}
let id = parts.next()?.strip_prefix("/hooks/")?.trim_end_matches('/').parse().ok()?;
let secret = head.lines().find_map(|l| {
let (k, v) = l.split_once(':')?;
k.trim().eq_ignore_ascii_case("authorization").then(|| v.trim().strip_prefix("Bearer ").unwrap_or("").to_string())
})?;
Some((id, secret, body.to_string()))
}
fn ensure_webhook_server(route: fn(u64, String)) -> Result<u16, String> {
if let Some(p) = webhook_port() {
return Ok(p);
}
let listener = (WEBHOOK_PORT..WEBHOOK_PORT + 20).find_map(|p| std::net::TcpListener::bind(("127.0.0.1", p)).ok()).ok_or("找不到可用的連接埠")?;
let port = listener.local_addr().map_err(|e| e.to_string())?.port();
*WEBHOOK_BOUND.lock().unwrap() = Some(port);
std::thread::spawn(move || {
for conn in listener.incoming().flatten() {
std::thread::spawn(move || {
let mut c = conn;
c.set_read_timeout(Some(Duration::from_secs(10))).ok();
let mut got = vec![];
let mut buf = [0u8; 8192];
loop {
let n = c.read(&mut buf).unwrap_or(0);
got.extend_from_slice(&buf[..n]);
let t = String::from_utf8_lossy(&got).to_string();
if let Some(h) = t.find("\r\n\r\n") {
let len = t[..h].lines().find_map(|l| l.to_lowercase().strip_prefix("content-length:").map(|v| v.trim().parse::<usize>().unwrap_or(0))).unwrap_or(0);
if got.len() >= h + 4 + len {
break;
}
}
if n == 0 || got.len() > 1 << 20 {
break;
}
}
let status = match parse_hook(&got) {
Some((id, secret, body)) => match webhooks().lock().unwrap().get(&id).cloned() {
Some((want, _)) if want == secret => {
route(id, format!("Webhook 被呼叫,內容:{}", body.chars().take(4000).collect::<String>()));
"202 Accepted"
}
Some(_) => "401 Unauthorized",
None => "404 Not Found",
},
None => "400 Bad Request",
};
let _ = c.write_all(format!("HTTP/1.1 {status}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n").as_bytes());
});
}
});
Ok(port)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn github_events_become_kinds() {
let v = json!([
{"id": "5", "type": "PushEvent", "actor": {"login": "amy"}, "payload": {"ref": "refs/heads/main", "size": 2}},
{"id": "3", "type": "PullRequestEvent", "actor": {"login": "bob"}, "payload": {"action": "closed", "pull_request": {"number": 7, "title": "Fix", "merged": true, "head": {"ref": "fix"}}}},
{"id": "4", "type": "PullRequestReviewEvent", "actor": {"login": "cat"}, "payload": {"review": {"state": "APPROVED", "body": "lgtm"}, "pull_request": {"number": 8, "head": {"ref": "x"}}}},
{"id": "2", "type": "IssuesEvent", "actor": {"login": "dan"}, "payload": {"action": "assigned", "issue": {"number": 9, "title": "Bug"}, "assignee": {"login": "me"}}},
{"id": "1", "type": "WatchEvent", "actor": {"login": "x"}, "payload": {}}
]);
let (cur, evs) = github_events(&v, 1);
assert_eq!(cur, 5);
assert_eq!(evs.iter().map(|e| e.kind.as_str()).collect::<Vec<_>>(), vec!["issue_assigned", "pr_merged", "review_approved", "push"]);
assert_eq!(evs[3].branch, "main");
assert_eq!(github_events(&v, 5).1.len(), 0);
}
#[test]
fn ci_runs_fire_once_after_the_first_look() {
let v = json!({"workflow_runs": [{"id": 1, "status": "completed", "conclusion": "failure", "name": "CI", "head_branch": "main"}]});
let mut seen = HashSet::new();
assert!(github_runs(&v, &mut seen, true).is_empty());
let v2 = json!({"workflow_runs": [{"id": 2, "status": "completed", "conclusion": "success", "name": "CI", "head_branch": "main"}, {"id": 1, "status": "completed", "conclusion": "failure"}]});
let e = github_runs(&v2, &mut seen, false);
assert_eq!(e.len(), 1);
assert_eq!(e[0].kind, "ci_passed");
}
#[test]
fn status_changes_find_new_and_changed() {
let mut prev = HashMap::new();
assert!(status_changes(&mut prev, &[("a".into(), "triggered".into(), "A".into())], true).is_empty());
let c = status_changes(&mut prev, &[("a".into(), "resolved".into(), "A".into()), ("b".into(), "triggered".into(), "B".into())], false);
assert_eq!(c, vec![("resolved".to_string(), "A(triggered → resolved)".to_string()), ("created".to_string(), "B".to_string())]);
}
#[test]
fn triggers_parse_and_validate() {
let t: Trigger = serde_json::from_value(json!({"type": "github", "repo": "me/app", "events": ["pr_merged", "ci_failed"]})).unwrap();
assert!(t.validate().is_ok());
assert!(t.describe().contains("me/app"));
let bad: Trigger = serde_json::from_value(json!({"type": "github", "repo": "app", "events": []})).unwrap();
assert!(bad.validate().is_err());
let g: Trigger = serde_json::from_value(json!({"type": "group", "triggers": [{"type": "webhook", "secret": "12345678"}]})).unwrap();
assert!(g.validate().is_err());
let s: Trigger = serde_json::from_value(json!({"type": "slack", "channel": "C1", "match": "keyword", "value": "部署"})).unwrap();
assert!(s.validate().is_ok() && slack_matches("今天要部署嗎", "keyword", "部署", "U1") && !slack_matches("hi", "mention", "", "U1"));
}
#[test]
fn calendar_fires_once_and_follows_moves() {
let now = chrono::DateTime::parse_from_rfc3339("2026-11-04T09:00:20+08:00").unwrap().timestamp();
let ev = |start: &str| json!([{"id": "E1", "title": "週會", "start": start}]);
let mut fired = HashSet::new();
assert_eq!(calendar_due(&ev("2026-11-04T09:00:00+08:00"), now, 0, 60, "", &mut fired).len(), 1);
assert!(calendar_due(&ev("2026-11-04T09:00:00+08:00"), now, 0, 60, "", &mut fired).is_empty());
// Moved to 10:00: fires again then, not now.
assert!(calendar_due(&ev("2026-11-04T10:00:00+08:00"), now, 0, 60, "", &mut fired).is_empty());
assert_eq!(calendar_due(&ev("2026-11-04T10:00:00+08:00"), now + 3600, 0, 60, "", &mut fired).len(), 1);
// 10 minutes before, filtered by title.
let mut f2 = HashSet::new();
assert_eq!(calendar_due(&ev("2026-11-04T09:10:00+08:00"), now, 10, 60, "週會", &mut f2).len(), 1);
assert!(calendar_due(&ev("2026-11-04T09:10:00+08:00"), now, 10, 60, "面試", &mut HashSet::new()).is_empty());
}
#[test]
fn webhook_needs_the_secret() {
let (tx, rx) = std::sync::mpsc::channel();
let watches = vec![Watch { bot_id: 3, routine_id: 41, trigger: Trigger::Webhook { secret: "s3cret-key".into() } }];
restart(watches, String::new(), tx, Arc::new(|| {}));
let port = webhook_port().unwrap();
let call = |auth: &str| {
let mut c = std::net::TcpStream::connect(("127.0.0.1", port)).unwrap();
let body = "{\"deploy\":\"ok\"}";
c.write_all(format!("POST /hooks/41 HTTP/1.1\r\nHost: x\r\nAuthorization: Bearer {auth}\r\nContent-Length: {}\r\n\r\n{body}", body.len()).as_bytes()).unwrap();
let mut out = String::new();
c.read_to_string(&mut out).unwrap();
out
};
assert!(call("wrong").starts_with("HTTP/1.1 401"));
assert!(call("s3cret-key").starts_with("HTTP/1.1 202"));
let f = rx.recv_timeout(Duration::from_secs(3)).unwrap();
assert_eq!((f.bot_id, f.routine_id), (3, 41));
assert!(f.what.contains("deploy"));
}
}