OXYGENHUB
# c96b32865000 oxygenhub / crates / hub / src / db / hooks.rs
10.6 KiB · Rust · last changed Webhooks: Discord and Slack formats (Discord hooks switched over), a masked URL that wraps, the last delivery in the table, an edit form, and web issue events delivered 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
    /// `oxygenhub`, `discord` or `slack`.
17
    pub format: String,
18
}
19
20
/// The formats a webhook can be delivered in.
21
pub const FORMATS: [&str; 3] = ["oxygenhub", "discord", "slack"];
22
23
/// The format a URL most likely wants: Discord and Slack webhooks accept
24
/// only their own bodies.
25
pub fn format_for_url(url: &str) -> &'static str {
26
    let host = url.split("://").nth(1).unwrap_or("").split(['/', ':']).next().unwrap_or("").to_ascii_lowercase();
27
    if matches!(host.as_str(), "discord.com" | "discordapp.com" | "canary.discord.com" | "ptb.discord.com")
28
        && url.contains("/api/webhooks/")
29
    {
30
        "discord"
31
    } else if host == "hooks.slack.com" {
32
        "slack"
33
    } else {
34
        "oxygenhub"
35
    }
36
}
37
38
/// A URL for display: scheme, host and path, with long segments (the tokens
39
/// Discord and Slack keep in the path) and any query cut to a few characters.
40
pub fn masked_url(url: &str) -> String {
41
    let (scheme, rest) = url.split_once("://").unwrap_or(("", url));
42
    let (path, query) = rest.split_once('?').map(|(p, q)| (p, Some(q))).unwrap_or((rest, None));
43
    let parts: Vec<String> =
44
        path.split('/')
45
            .enumerate()
46
            .map(|(i, seg)| {
47
                if i > 0 && seg.chars().count() > 14 {
48
                    format!("{}…", seg.chars().take(4).collect::<String>())
49
                } else {
50
                    seg.to_string()
51
                }
52
            })
53
            .collect();
54
    let mut out = if scheme.is_empty() { parts.join("/") } else { format!("{scheme}://{}", parts.join("/")) };
55
    if query.is_some() {
56
        out.push_str("?…");
57
    }
58
    out
59
}
60
61
impl Hook {
62
    /// Whether the hook wants `event` (`*`, or a comma list of prefixes such as `push,proposal`).
63
    pub fn wants(&self, event: &str) -> bool {
64
        self.active && self.events.split(',').map(str::trim).any(|e| e == "*" || event == e || event.starts_with(&format!("{e}.")))
65
    }
66
}
67
68
#[derive(Clone, Debug, FromRow)]
69
pub struct Delivery {
70
    pub id: i64,
71
    pub hook_id: i64,
72
    pub delivery: String,
73
    pub event: String,
74
    pub payload: String,
75
    pub created_ms: i64,
76
    pub attempts: i64,
77
    pub next_ms: i64,
78
    pub status: Option<i64>,
79
    pub response: String,
80
    pub delivered_ms: Option<i64>,
81
}
82
83
pub async fn hooks_for_repo(pool: &Pool, repo_id: i64) -> sqlx::Result<Vec<Hook>> {
84
    sqlx::query_as("SELECT * FROM hooks WHERE repo_id = ? ORDER BY id").bind(repo_id).fetch_all(pool).await
85
}
86
87
pub async fn hooks_for_org(pool: &Pool, org_id: i64) -> sqlx::Result<Vec<Hook>> {
88
    sqlx::query_as("SELECT * FROM hooks WHERE org_id = ? ORDER BY id").bind(org_id).fetch_all(pool).await
89
}
90
91
/// Every hook that should hear about an event in a repository: its own
92
/// and its organisation's.
93
pub async fn hooks_hearing(pool: &Pool, repo_id: i64, owner: &str) -> sqlx::Result<Vec<Hook>> {
94
    let mut out = hooks_for_repo(pool, repo_id).await?;
95
    if let Some(org) = super::accounts::org_by_name(pool, owner).await? {
96
        out.extend(hooks_for_org(pool, org.id).await?);
97
    }
98
    Ok(out)
99
}
100
101
pub async fn hook(pool: &Pool, id: i64) -> sqlx::Result<Option<Hook>> {
102
    sqlx::query_as("SELECT * FROM hooks WHERE id = ?").bind(id).fetch_optional(pool).await
103
}
104
105
pub async fn create_hook(
106
    pool: &Pool,
107
    repo_id: Option<i64>,
108
    org_id: Option<i64>,
109
    url: &str,
110
    secret: &str,
111
    events: &str,
112
) -> sqlx::Result<Hook> {
113
    sqlx::query_as("INSERT INTO hooks (repo_id, org_id, url, secret, events, created_ms, format) VALUES (?, ?, ?, ?, ?, ?, ?) RETURNING *")
114
        .bind(repo_id)
115
        .bind(org_id)
116
        .bind(url)
117
        .bind(secret)
118
        .bind(events)
119
        .bind(now_ms())
120
        .bind(format_for_url(url))
121
        .fetch_one(pool)
122
        .await
123
}
124
125
/// Change a hook; `None` keeps the URL or secret as it is.
126
pub async fn update_hook(
127
    pool: &Pool,
128
    id: i64,
129
    url: Option<&str>,
130
    secret: Option<&str>,
131
    events: &str,
132
    format: &str,
133
    active: bool,
134
) -> sqlx::Result<()> {
135
    sqlx::query("UPDATE hooks SET url = COALESCE(?, url), secret = COALESCE(?, secret), events = ?, format = ?, active = ? WHERE id = ?")
136
        .bind(url)
137
        .bind(secret)
138
        .bind(events)
139
        .bind(format)
140
        .bind(active)
141
        .bind(id)
142
        .execute(pool)
143
        .await?;
144
    Ok(())
145
}
146
147
pub async fn set_hook_format(pool: &Pool, id: i64, format: &str) -> sqlx::Result<()> {
148
    sqlx::query("UPDATE hooks SET format = ? WHERE id = ?").bind(format).bind(id).execute(pool).await?;
149
    Ok(())
150
}
151
152
/// The newest delivery that was attempted, for the settings table.
153
pub async fn last_attempt(pool: &Pool, hook_id: i64) -> sqlx::Result<Option<Delivery>> {
154
    sqlx::query_as("SELECT * FROM deliveries WHERE hook_id = ? AND attempts > 0 ORDER BY id DESC LIMIT 1")
155
        .bind(hook_id)
156
        .fetch_optional(pool)
157
        .await
158
}
159
160
pub async fn set_hook_active(pool: &Pool, id: i64, active: bool) -> sqlx::Result<()> {
161
    sqlx::query("UPDATE hooks SET active = ? WHERE id = ?").bind(active).bind(id).execute(pool).await?;
162
    Ok(())
163
}
164
165
pub async fn delete_hook(pool: &Pool, id: i64) -> sqlx::Result<()> {
166
    sqlx::query("DELETE FROM hooks WHERE id = ?").bind(id).execute(pool).await?;
167
    Ok(())
168
}
169
170
pub async fn queue_delivery(pool: &Pool, hook_id: i64, delivery: &str, event: &str, payload: &str) -> sqlx::Result<i64> {
171
    let now = now_ms();
172
    sqlx::query_scalar(
173
        "INSERT INTO deliveries (hook_id, delivery, event, payload, created_ms, next_ms) VALUES (?, ?, ?, ?, ?, ?) RETURNING id",
174
    )
175
    .bind(hook_id)
176
    .bind(delivery)
177
    .bind(event)
178
    .bind(payload)
179
    .bind(now)
180
    .bind(now)
181
    .fetch_one(pool)
182
    .await
183
}
184
185
pub async fn due_deliveries(pool: &Pool, limit: i64) -> sqlx::Result<Vec<Delivery>> {
186
    sqlx::query_as("SELECT * FROM deliveries WHERE delivered_ms IS NULL AND next_ms <= ? ORDER BY next_ms LIMIT ?")
187
        .bind(now_ms())
188
        .bind(limit)
189
        .fetch_all(pool)
190
        .await
191
}
192
193
pub async fn deliveries_for_hook(pool: &Pool, hook_id: i64, limit: i64) -> sqlx::Result<Vec<Delivery>> {
194
    sqlx::query_as("SELECT * FROM deliveries WHERE hook_id = ? ORDER BY created_ms DESC, id DESC LIMIT ?")
195
        .bind(hook_id)
196
        .bind(limit)
197
        .fetch_all(pool)
198
        .await
199
}
200
201
pub async fn delivery(pool: &Pool, id: i64) -> sqlx::Result<Option<Delivery>> {
202
    sqlx::query_as("SELECT * FROM deliveries WHERE id = ?").bind(id).fetch_optional(pool).await
203
}
204
205
/// Record an attempt: done on a 2xx, else retried with backoff up to a day.
206
pub async fn record_attempt(pool: &Pool, id: i64, attempts: i64, status: Option<i64>, response: &str) -> sqlx::Result<()> {
207
    let ok = status.is_some_and(|s| (200..300).contains(&s));
208
    let now = now_ms();
209
    let delivered: Option<i64> = if ok { Some(now) } else { None };
210
    // 1 min, 5 min, 30 min, 2 h, 6 h, then give up after a day's worth.
211
    let backoff_ms = [60_000i64, 300_000, 1_800_000, 7_200_000, 21_600_000];
212
    let next = if ok || attempts as usize >= backoff_ms.len() { i64::MAX / 4 } else { now + backoff_ms[attempts as usize] };
213
    sqlx::query("UPDATE deliveries SET attempts = ?, status = ?, response = ?, delivered_ms = ?, next_ms = ? WHERE id = ?")
214
        .bind(attempts + 1)
215
        .bind(status)
216
        .bind(&response.chars().take(2000).collect::<String>())
217
        .bind(delivered)
218
        .bind(next)
219
        .bind(id)
220
        .execute(pool)
221
        .await?;
222
    Ok(())
223
}
224
225
/// Requeue a delivery for an immediate retry.
226
pub async fn redeliver(pool: &Pool, id: i64) -> sqlx::Result<()> {
227
    sqlx::query("UPDATE deliveries SET delivered_ms = NULL, next_ms = ? WHERE id = ?").bind(now_ms()).bind(id).execute(pool).await?;
228
    Ok(())
229
}
230
231
pub async fn prune_deliveries(pool: &Pool, older_than_ms: i64) -> sqlx::Result<()> {
232
    sqlx::query("DELETE FROM deliveries WHERE created_ms < ?").bind(older_than_ms).execute(pool).await?;
233
    Ok(())
234
}
235
236
// ------------------------------------------------------------------ checks
237
238
#[derive(Clone, Debug, FromRow)]
239
pub struct Check {
240
    pub id: i64,
241
    pub repo_id: i64,
242
    pub state: String,
243
    pub name: String,
244
    pub status: String,
245
    pub summary: String,
246
    pub url: String,
247
    pub reporter_id: Option<i64>,
248
    pub reporter: String,
249
    pub seal: String,
250
    pub started_ms: i64,
251
    pub finished_ms: Option<i64>,
252
}
253
254
impl Check {
255
    pub fn done(&self) -> bool {
256
        matches!(self.status.as_str(), "passed" | "failed" | "cancelled" | "skipped")
257
    }
258
}
259
260
pub async fn checks_for_state(pool: &Pool, repo_id: i64, state: &str) -> sqlx::Result<Vec<Check>> {
261
    sqlx::query_as("SELECT * FROM checks WHERE repo_id = ? AND state = ? ORDER BY name").bind(repo_id).bind(state).fetch_all(pool).await
262
}
263
264
/// Create or update a check by `(state, name)`.
265
pub async fn upsert_check(
266
    pool: &Pool,
267
    repo_id: i64,
268
    state: &str,
269
    name: &str,
270
    status: &str,
271
    summary: &str,
272
    url: &str,
273
    reporter: Option<(i64, &str)>,
274
    seal: &str,
275
) -> sqlx::Result<Check> {
276
    let now = now_ms();
277
    let finished: Option<i64> = if matches!(status, "passed" | "failed" | "cancelled" | "skipped") { Some(now) } else { None };
278
    sqlx::query_as(
279
        "INSERT INTO checks (repo_id, state, name, status, summary, url, reporter_id, reporter, seal, started_ms, finished_ms) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) \
280
         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 *",
281
    )
282
    .bind(repo_id)
283
    .bind(state)
284
    .bind(name)
285
    .bind(status)
286
    .bind(summary)
287
    .bind(url)
288
    .bind(reporter.map(|r| r.0))
289
    .bind(reporter.map(|r| r.1).unwrap_or(""))
290
    .bind(seal)
291
    .bind(now)
292
    .bind(finished)
293
    .fetch_one(pool)
294
    .await
295
}
296
297
pub async fn required_checks(pool: &Pool, repo_id: i64) -> sqlx::Result<Vec<String>> {
298
    let s: String = sqlx::query_scalar("SELECT required_checks FROM repositories WHERE id = ?").bind(repo_id).fetch_one(pool).await?;
299
    Ok(s.split(',').map(str::trim).filter(|x| !x.is_empty()).map(String::from).collect())
300
}
301
302
pub async fn set_required_checks(pool: &Pool, repo_id: i64, names: &[String]) -> sqlx::Result<()> {
303
    sqlx::query("UPDATE repositories SET required_checks = ? WHERE id = ?").bind(names.join(",")).bind(repo_id).execute(pool).await?;
304
    Ok(())
305
}