OXYGENHUB
# 56a786aa42f1 oxygenhub / crates / hub / src / db / hooks.rs
7.9 KiB · Rust · last changed Milestone 0.5: API, webhooks, checks 1 month ago source blame history raw
1 
//! Webhooks, their delivery log, and checks (hub spec §12.2, §13.1).
2
3 
use super::{now_ms, Pool};
4 
use sqlx::FromRow;
5
6 
#[derive(Clone, Debug, FromRow)]
7 
pub struct Hook {
8 
    pub id: i64,
9 
    pub repo_id: Option<i64>,
10 
    pub org_id: Option<i64>,
11 
    pub url: String,
12 
    pub secret: String,
13 
    pub events: String,
14 
    pub active: bool,
15 
    pub created_ms: i64,
16 
}
17
18 
impl Hook {
19 
    /// Whether the hook wants `event` (`*`, or a comma list of prefixes such as `push,proposal`).
20 
    pub fn wants(&self, event: &str) -> bool {
21 
        self.active && self.events.split(',').map(str::trim).any(|e| e == "*" || event == e || event.starts_with(&format!("{e}.")))
22 
    }
23 
}
24
25 
#[derive(Clone, Debug, FromRow)]
26 
pub struct Delivery {
27 
    pub id: i64,
28 
    pub hook_id: i64,
29 
    pub delivery: String,
30 
    pub event: String,
31 
    pub payload: String,
32 
    pub created_ms: i64,
33 
    pub attempts: i64,
34 
    pub next_ms: i64,
35 
    pub status: Option<i64>,
36 
    pub response: String,
37 
    pub delivered_ms: Option<i64>,
38 
}
39
40 
pub async fn hooks_for_repo(pool: &Pool, repo_id: i64) -> sqlx::Result<Vec<Hook>> {
41 
    sqlx::query_as("SELECT * FROM hooks WHERE repo_id = ? ORDER BY id").bind(repo_id).fetch_all(pool).await
42 
}
43
44 
pub async fn hooks_for_org(pool: &Pool, org_id: i64) -> sqlx::Result<Vec<Hook>> {
45 
    sqlx::query_as("SELECT * FROM hooks WHERE org_id = ? ORDER BY id").bind(org_id).fetch_all(pool).await
46 
}
47
48 
/// Every hook that should hear about an event in a repository: its own
49 
/// and its organisation's.
50 
pub async fn hooks_hearing(pool: &Pool, repo_id: i64, owner: &str) -> sqlx::Result<Vec<Hook>> {
51 
    let mut out = hooks_for_repo(pool, repo_id).await?;
52 
    if let Some(org) = super::accounts::org_by_name(pool, owner).await? {
53 
        out.extend(hooks_for_org(pool, org.id).await?);
54 
    }
55 
    Ok(out)
56 
}
57
58 
pub async fn hook(pool: &Pool, id: i64) -> sqlx::Result<Option<Hook>> {
59 
    sqlx::query_as("SELECT * FROM hooks WHERE id = ?").bind(id).fetch_optional(pool).await
60 
}
61
62 
pub async fn create_hook(
63 
    pool: &Pool,
64 
    repo_id: Option<i64>,
65 
    org_id: Option<i64>,
66 
    url: &str,
67 
    secret: &str,
68 
    events: &str,
69 
) -> sqlx::Result<Hook> {
70 
    sqlx::query_as("INSERT INTO hooks (repo_id, org_id, url, secret, events, created_ms) VALUES (?, ?, ?, ?, ?, ?) RETURNING *")
71 
        .bind(repo_id)
72 
        .bind(org_id)
73 
        .bind(url)
74 
        .bind(secret)
75 
        .bind(events)
76 
        .bind(now_ms())
77 
        .fetch_one(pool)
78 
        .await
79 
}
80
81 
pub async fn set_hook_active(pool: &Pool, id: i64, active: bool) -> sqlx::Result<()> {
82 
    sqlx::query("UPDATE hooks SET active = ? WHERE id = ?").bind(active).bind(id).execute(pool).await?;
83 
    Ok(())
84 
}
85
86 
pub async fn delete_hook(pool: &Pool, id: i64) -> sqlx::Result<()> {
87 
    sqlx::query("DELETE FROM hooks WHERE id = ?").bind(id).execute(pool).await?;
88 
    Ok(())
89 
}
90
91 
pub async fn queue_delivery(pool: &Pool, hook_id: i64, delivery: &str, event: &str, payload: &str) -> sqlx::Result<i64> {
92 
    let now = now_ms();
93 
    sqlx::query_scalar(
94 
        "INSERT INTO deliveries (hook_id, delivery, event, payload, created_ms, next_ms) VALUES (?, ?, ?, ?, ?, ?) RETURNING id",
95 
    )
96 
    .bind(hook_id)
97 
    .bind(delivery)
98 
    .bind(event)
99 
    .bind(payload)
100 
    .bind(now)
101 
    .bind(now)
102 
    .fetch_one(pool)
103 
    .await
104 
}
105
106 
pub async fn due_deliveries(pool: &Pool, limit: i64) -> sqlx::Result<Vec<Delivery>> {
107 
    sqlx::query_as("SELECT * FROM deliveries WHERE delivered_ms IS NULL AND next_ms <= ? ORDER BY next_ms LIMIT ?")
108 
        .bind(now_ms())
109 
        .bind(limit)
110 
        .fetch_all(pool)
111 
        .await
112 
}
113
114 
pub async fn deliveries_for_hook(pool: &Pool, hook_id: i64, limit: i64) -> sqlx::Result<Vec<Delivery>> {
115 
    sqlx::query_as("SELECT * FROM deliveries WHERE hook_id = ? ORDER BY created_ms DESC, id DESC LIMIT ?")
116 
        .bind(hook_id)
117 
        .bind(limit)
118 
        .fetch_all(pool)
119 
        .await
120 
}
121
122 
pub async fn delivery(pool: &Pool, id: i64) -> sqlx::Result<Option<Delivery>> {
123 
    sqlx::query_as("SELECT * FROM deliveries WHERE id = ?").bind(id).fetch_optional(pool).await
124 
}
125
126 
/// Record an attempt: done on a 2xx, else retried with backoff up to a day.
127 
pub async fn record_attempt(pool: &Pool, id: i64, attempts: i64, status: Option<i64>, response: &str) -> sqlx::Result<()> {
128 
    let ok = status.is_some_and(|s| (200..300).contains(&s));
129 
    let now = now_ms();
130 
    let delivered: Option<i64> = if ok { Some(now) } else { None };
131 
    // 1 min, 5 min, 30 min, 2 h, 6 h, then give up after a day's worth.
132 
    let backoff_ms = [60_000i64, 300_000, 1_800_000, 7_200_000, 21_600_000];
133 
    let next = if ok || attempts as usize >= backoff_ms.len() { i64::MAX / 4 } else { now + backoff_ms[attempts as usize] };
134 
    sqlx::query("UPDATE deliveries SET attempts = ?, status = ?, response = ?, delivered_ms = ?, next_ms = ? WHERE id = ?")
135 
        .bind(attempts + 1)
136 
        .bind(status)
137 
        .bind(&response.chars().take(2000).collect::<String>())
138 
        .bind(delivered)
139 
        .bind(next)
140 
        .bind(id)
141 
        .execute(pool)
142 
        .await?;
143 
    Ok(())
144 
}
145
146 
/// Requeue a delivery for an immediate retry.
147 
pub async fn redeliver(pool: &Pool, id: i64) -> sqlx::Result<()> {
148 
    sqlx::query("UPDATE deliveries SET delivered_ms = NULL, next_ms = ? WHERE id = ?").bind(now_ms()).bind(id).execute(pool).await?;
149 
    Ok(())
150 
}
151
152 
pub async fn prune_deliveries(pool: &Pool, older_than_ms: i64) -> sqlx::Result<()> {
153 
    sqlx::query("DELETE FROM deliveries WHERE created_ms < ?").bind(older_than_ms).execute(pool).await?;
154 
    Ok(())
155 
}
156
157 
// ------------------------------------------------------------------ checks
158
159 
#[derive(Clone, Debug, FromRow)]
160 
pub struct Check {
161 
    pub id: i64,
162 
    pub repo_id: i64,
163 
    pub state: String,
164 
    pub name: String,
165 
    pub status: String,
166 
    pub summary: String,
167 
    pub url: String,
168 
    pub reporter_id: Option<i64>,
169 
    pub reporter: String,
170 
    pub seal: String,
171 
    pub started_ms: i64,
172 
    pub finished_ms: Option<i64>,
173 
}
174
175 
impl Check {
176 
    pub fn done(&self) -> bool {
177 
        matches!(self.status.as_str(), "passed" | "failed" | "cancelled" | "skipped")
178 
    }
179 
}
180
181 
pub async fn checks_for_state(pool: &Pool, repo_id: i64, state: &str) -> sqlx::Result<Vec<Check>> {
182 
    sqlx::query_as("SELECT * FROM checks WHERE repo_id = ? AND state = ? ORDER BY name").bind(repo_id).bind(state).fetch_all(pool).await
183 
}
184
185 
/// Create or update a check by `(state, name)`.
186 
pub async fn upsert_check(
187 
    pool: &Pool,
188 
    repo_id: i64,
189 
    state: &str,
190 
    name: &str,
191 
    status: &str,
192 
    summary: &str,
193 
    url: &str,
194 
    reporter: Option<(i64, &str)>,
195 
    seal: &str,
196 
) -> sqlx::Result<Check> {
197 
    let now = now_ms();
198 
    let finished: Option<i64> = if matches!(status, "passed" | "failed" | "cancelled" | "skipped") { Some(now) } else { None };
199 
    sqlx::query_as(
200 
        "INSERT INTO checks (repo_id, state, name, status, summary, url, reporter_id, reporter, seal, started_ms, finished_ms) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) \
201 
         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 *",
202 
    )
203 
    .bind(repo_id)
204 
    .bind(state)
205 
    .bind(name)
206 
    .bind(status)
207 
    .bind(summary)
208 
    .bind(url)
209 
    .bind(reporter.map(|r| r.0))
210 
    .bind(reporter.map(|r| r.1).unwrap_or(""))
211 
    .bind(seal)
212 
    .bind(now)
213 
    .bind(finished)
214 
    .fetch_one(pool)
215 
    .await
216 
}
217
218 
pub async fn required_checks(pool: &Pool, repo_id: i64) -> sqlx::Result<Vec<String>> {
219 
    let s: String = sqlx::query_scalar("SELECT required_checks FROM repositories WHERE id = ?").bind(repo_id).fetch_one(pool).await?;
220 
    Ok(s.split(',').map(str::trim).filter(|x| !x.is_empty()).map(String::from).collect())
221 
}
222
223 
pub async fn set_required_checks(pool: &Pool, repo_id: i64, names: &[String]) -> sqlx::Result<()> {
224 
    sqlx::query("UPDATE repositories SET required_checks = ? WHERE id = ?").bind(names.join(",")).bind(repo_id).execute(pool).await?;
225 
    Ok(())
226 
}