//! Webhooks, their delivery log, and checks (hub spec §12.2, §13.1). use super::{now_ms, Pool}; use sqlx::FromRow; #[derive(Clone, Debug, FromRow)] pub struct Hook { pub id: i64, pub repo_id: Option, pub org_id: Option, pub url: String, pub secret: String, pub events: String, pub active: bool, pub created_ms: i64, /// `oxygenhub`, `discord` or `slack`. pub format: String, } /// The formats a webhook can be delivered in. pub const FORMATS: [&str; 3] = ["oxygenhub", "discord", "slack"]; /// The format a URL most likely wants: Discord and Slack webhooks accept /// only their own bodies. pub fn format_for_url(url: &str) -> &'static str { let host = url.split("://").nth(1).unwrap_or("").split(['/', ':']).next().unwrap_or("").to_ascii_lowercase(); if matches!(host.as_str(), "discord.com" | "discordapp.com" | "canary.discord.com" | "ptb.discord.com") && url.contains("/api/webhooks/") { "discord" } else if host == "hooks.slack.com" { "slack" } else { "oxygenhub" } } /// A URL for display: scheme, host and path, with long segments (the tokens /// Discord and Slack keep in the path) and any query cut to a few characters. pub fn masked_url(url: &str) -> String { let (scheme, rest) = url.split_once("://").unwrap_or(("", url)); let (path, query) = rest.split_once('?').map(|(p, q)| (p, Some(q))).unwrap_or((rest, None)); let parts: Vec = path.split('/') .enumerate() .map(|(i, seg)| { if i > 0 && seg.chars().count() > 14 { format!("{}…", seg.chars().take(4).collect::()) } else { seg.to_string() } }) .collect(); let mut out = if scheme.is_empty() { parts.join("/") } else { format!("{scheme}://{}", parts.join("/")) }; if query.is_some() { out.push_str("?…"); } out } impl Hook { /// Whether the hook wants `event` (`*`, or a comma list of prefixes such as `push,proposal`). pub fn wants(&self, event: &str) -> bool { self.active && self.events.split(',').map(str::trim).any(|e| e == "*" || event == e || event.starts_with(&format!("{e}."))) } } #[derive(Clone, Debug, FromRow)] pub struct Delivery { pub id: i64, pub hook_id: i64, pub delivery: String, pub event: String, pub payload: String, pub created_ms: i64, pub attempts: i64, pub next_ms: i64, pub status: Option, pub response: String, pub delivered_ms: Option, } pub async fn hooks_for_repo(pool: &Pool, repo_id: i64) -> sqlx::Result> { sqlx::query_as("SELECT * FROM hooks WHERE repo_id = $1 ORDER BY id").bind(repo_id).fetch_all(pool).await } pub async fn hooks_for_org(pool: &Pool, org_id: i64) -> sqlx::Result> { sqlx::query_as("SELECT * FROM hooks WHERE org_id = $1 ORDER BY id").bind(org_id).fetch_all(pool).await } /// Every hook that should hear about an event in a repository: its own /// and its organisation's. pub async fn hooks_hearing(pool: &Pool, repo_id: i64, owner: &str) -> sqlx::Result> { let mut out = hooks_for_repo(pool, repo_id).await?; if let Some(org) = super::accounts::org_by_name(pool, owner).await? { out.extend(hooks_for_org(pool, org.id).await?); } Ok(out) } pub async fn hook(pool: &Pool, id: i64) -> sqlx::Result> { sqlx::query_as("SELECT * FROM hooks WHERE id = $1").bind(id).fetch_optional(pool).await } pub async fn create_hook( pool: &Pool, repo_id: Option, org_id: Option, url: &str, secret: &str, events: &str, ) -> sqlx::Result { sqlx::query_as( "INSERT INTO hooks (repo_id, org_id, url, secret, events, created_ms, format) VALUES ($1, $2, $3, $4, $5, $6, $7) RETURNING *", ) .bind(repo_id) .bind(org_id) .bind(url) .bind(secret) .bind(events) .bind(now_ms()) .bind(format_for_url(url)) .fetch_one(pool) .await } /// Change a hook; `None` keeps the URL or secret as it is. pub async fn update_hook( pool: &Pool, id: i64, url: Option<&str>, secret: Option<&str>, events: &str, format: &str, active: bool, ) -> sqlx::Result<()> { sqlx::query( "UPDATE hooks SET url = COALESCE($1, url), secret = COALESCE($2, secret), events = $3, format = $4, active = $5 WHERE id = $6", ) .bind(url) .bind(secret) .bind(events) .bind(format) .bind(active) .bind(id) .execute(pool) .await?; Ok(()) } pub async fn set_hook_format(pool: &Pool, id: i64, format: &str) -> sqlx::Result<()> { sqlx::query("UPDATE hooks SET format = $1 WHERE id = $2").bind(format).bind(id).execute(pool).await?; Ok(()) } /// The newest delivery that was attempted, for the settings table. pub async fn last_attempt(pool: &Pool, hook_id: i64) -> sqlx::Result> { sqlx::query_as("SELECT * FROM deliveries WHERE hook_id = $1 AND attempts > 0 ORDER BY id DESC LIMIT 1") .bind(hook_id) .fetch_optional(pool) .await } pub async fn set_hook_active(pool: &Pool, id: i64, active: bool) -> sqlx::Result<()> { sqlx::query("UPDATE hooks SET active = $1 WHERE id = $2").bind(active).bind(id).execute(pool).await?; Ok(()) } pub async fn delete_hook(pool: &Pool, id: i64) -> sqlx::Result<()> { sqlx::query("DELETE FROM hooks WHERE id = $1").bind(id).execute(pool).await?; Ok(()) } /// The most deliveries one hook may have waiting (#220): beyond it, new /// events for that hook are dropped (and logged) until it catches up. pub const MAX_WAITING: i64 = 1000; /// Queue a delivery; `Ok(None)` when the hook already has /// [`MAX_WAITING`] deliveries waiting, so nothing was queued. pub async fn queue_delivery(pool: &Pool, hook_id: i64, delivery: &str, event: &str, payload: &str) -> sqlx::Result> { let waiting: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM deliveries WHERE hook_id = $1 AND delivered_ms IS NULL AND next_ms < $2") .bind(hook_id) .bind(i64::MAX / 4) .fetch_one(pool) .await?; if waiting >= MAX_WAITING { return Ok(None); } let now = now_ms(); sqlx::query_scalar( "INSERT INTO deliveries (hook_id, delivery, event, payload, created_ms, next_ms) VALUES ($1, $2, $3, $4, $5, $6) RETURNING id", ) .bind(hook_id) .bind(delivery) .bind(event) .bind(payload) .bind(now) .bind(now) .fetch_one(pool) .await .map(Some) } pub async fn due_deliveries(pool: &Pool, limit: i64) -> sqlx::Result> { sqlx::query_as("SELECT * FROM deliveries WHERE delivered_ms IS NULL AND next_ms <= $1 ORDER BY next_ms LIMIT $2") .bind(now_ms()) .bind(limit) .fetch_all(pool) .await } pub async fn deliveries_for_hook(pool: &Pool, hook_id: i64, limit: i64) -> sqlx::Result> { sqlx::query_as("SELECT * FROM deliveries WHERE hook_id = $1 ORDER BY created_ms DESC, id DESC LIMIT $2") .bind(hook_id) .bind(limit) .fetch_all(pool) .await } pub async fn delivery(pool: &Pool, id: i64) -> sqlx::Result> { sqlx::query_as("SELECT * FROM deliveries WHERE id = $1").bind(id).fetch_optional(pool).await } /// Record an attempt: done on a 2xx, else retried with backoff up to a day. pub async fn record_attempt(pool: &Pool, id: i64, attempts: i64, status: Option, response: &str) -> sqlx::Result<()> { let ok = status.is_some_and(|s| (200..300).contains(&s)); let now = now_ms(); let delivered: Option = if ok { Some(now) } else { None }; // 1 min, 5 min, 30 min, 2 h, 6 h, then give up after a day's worth. let backoff_ms = [60_000i64, 300_000, 1_800_000, 7_200_000, 21_600_000]; let next = if ok || attempts as usize >= backoff_ms.len() { i64::MAX / 4 } else { now + backoff_ms[attempts as usize] }; sqlx::query("UPDATE deliveries SET attempts = $1, status = $2, response = $3, delivered_ms = $4, next_ms = $5 WHERE id = $6") .bind(attempts + 1) .bind(status) .bind(response.chars().take(2000).collect::()) .bind(delivered) .bind(next) .bind(id) .execute(pool) .await?; Ok(()) } /// Requeue a delivery for an immediate retry. pub async fn redeliver(pool: &Pool, id: i64) -> sqlx::Result<()> { sqlx::query("UPDATE deliveries SET delivered_ms = NULL, next_ms = $1 WHERE id = $2").bind(now_ms()).bind(id).execute(pool).await?; Ok(()) } pub async fn prune_deliveries(pool: &Pool, older_than_ms: i64) -> sqlx::Result<()> { sqlx::query("DELETE FROM deliveries WHERE created_ms < $1").bind(older_than_ms).execute(pool).await?; Ok(()) } // ------------------------------------------------------------------ checks #[derive(Clone, Debug, FromRow)] pub struct Check { pub id: i64, pub repo_id: i64, pub state: String, pub name: String, pub status: String, pub summary: String, pub url: String, pub reporter_id: Option, pub reporter: String, pub seal: String, pub started_ms: i64, pub finished_ms: Option, } impl Check { pub fn done(&self) -> bool { matches!(self.status.as_str(), "passed" | "failed" | "cancelled" | "skipped") } } pub async fn checks_for_state(pool: &Pool, repo_id: i64, state: &str) -> sqlx::Result> { sqlx::query_as("SELECT * FROM checks WHERE repo_id = $1 AND state = $2 ORDER BY name").bind(repo_id).bind(state).fetch_all(pool).await } /// Create or update a check by `(state, name)`. /// A check as a reporter posts it. pub struct CheckReport<'a> { pub state: &'a str, pub name: &'a str, pub status: &'a str, pub summary: &'a str, pub url: &'a str, pub reporter: Option<(i64, &'a str)>, pub seal: &'a str, } pub async fn upsert_check(pool: &Pool, repo_id: i64, c: CheckReport<'_>) -> sqlx::Result { let CheckReport { state, name, status, summary, url, reporter, seal } = c; let now = now_ms(); let finished: Option = if matches!(status, "passed" | "failed" | "cancelled" | "skipped") { Some(now) } else { None }; sqlx::query_as( "INSERT INTO checks (repo_id, state, name, status, summary, url, reporter_id, reporter, seal, started_ms, finished_ms) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) \ ON CONFLICT (repo_id, state, name) DO UPDATE SET status = excluded.status, summary = excluded.summary, url = excluded.url, reporter_id = excluded.reporter_id, reporter = excluded.reporter, seal = CASE WHEN excluded.seal = '' THEN checks.seal ELSE excluded.seal END, finished_ms = excluded.finished_ms RETURNING *", ) .bind(repo_id) .bind(state) .bind(name) .bind(status) .bind(summary) .bind(url) .bind(reporter.map(|r| r.0)) .bind(reporter.map(|r| r.1).unwrap_or("")) .bind(seal) .bind(now) .bind(finished) .fetch_one(pool) .await } pub async fn required_checks(pool: &Pool, repo_id: i64) -> sqlx::Result> { let s: String = sqlx::query_scalar("SELECT required_checks FROM repositories WHERE id = $1").bind(repo_id).fetch_one(pool).await?; Ok(s.split(',').map(str::trim).filter(|x| !x.is_empty()).map(String::from).collect()) } pub async fn set_required_checks(pool: &Pool, repo_id: i64, names: &[String]) -> sqlx::Result<()> { sqlx::query("UPDATE repositories SET required_checks = $1 WHERE id = $2").bind(names.join(",")).bind(repo_id).execute(pool).await?; Ok(()) }