883 lines
45 KiB
Rust
883 lines
45 KiB
Rust
// Event-triggered routines (trigger kinds, following the 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"));
|
||
}
|
||
}
|