| 1 | //! The webhooks service: events, delivered to the addresses a repository or |
| 2 | //! a workspace registers. See `g1t_contracts::webhooks` for the methods and |
| 3 | //! their arguments. |
| 4 | //! |
| 5 | //! An event from the bus becomes a delivery for each active webhook that |
| 6 | //! wants it, and is sent at once. One that is not answered with a 2xx is |
| 7 | //! tried again by the minute's sweep, waiting longer each time, until it |
| 8 | //! has had every attempt. Every delivery is kept for a fortnight, with what |
| 9 | //! was sent and what came back. |
| 10 | |
| 11 | mod deliver; |
| 12 | mod rename; |
| 13 | |
| 14 | use std::time::Duration; |
| 15 | |
| 16 | use futures_util::future::{Either, select}; |
| 17 | use g1t_contracts::access::{self, Capability}; |
| 18 | use g1t_contracts::events::Event; |
| 19 | use g1t_contracts::identity::{AGENT_ID, AGENT_NAME, UsernamesArgs}; |
| 20 | use g1t_contracts::repos::{GetArgs, GetByIdArgs, Repo}; |
| 21 | use g1t_contracts::time::rfc3339; |
| 22 | use g1t_contracts::webhooks::*; |
| 23 | use g1t_contracts::{FailureCode, Membership, Outcome, PrincipalKind, Role, User, Viewer, new_id}; |
| 24 | use g1t_kit::{args, now_ms, reply, rpc_method}; |
| 25 | use g1t_secrets::Sealer; |
| 26 | use serde::Deserialize; |
| 27 | use serde_json::Value; |
| 28 | use worker::wasm_bindgen::JsValue; |
| 29 | use worker::{ |
| 30 | Context, D1Database, Delay, Env, Fetch, Fetcher, Headers, MessageBatch, MessageExt, Method, Request, RequestInit, Response, |
| 31 | Result, ScheduleContext, ScheduledEvent, event, |
| 32 | }; |
| 33 | |
| 34 | /// How long a receiver has to answer. |
| 35 | const TIMEOUT: Duration = Duration::from_secs(10); |
| 36 | /// How much of an answer is kept. |
| 37 | const RESPONSE_KEPT: usize = 2_000; |
| 38 | const DELIVERIES_SHOWN: u32 = 50; |
| 39 | /// How long deliveries are kept. |
| 40 | const KEPT_DAYS: u64 = 14; |
| 41 | /// How many due retries one sweep makes. |
| 42 | const SWEEP: u32 = 50; |
| 43 | /// The hourly cron that forgets old deliveries (wrangler.jsonc); every |
| 44 | /// other minute only retries. |
| 45 | const PURGE_CRON: &str = "37 * * * *"; |
| 46 | /// How many old deliveries one statement deletes, and how many statements |
| 47 | /// one purge makes: a busy hour's worth, in bites D1 answers quickly. |
| 48 | const PURGE_BATCH: u32 = 1_000; |
| 49 | const PURGE_BATCHES: u32 = 50; |
| 50 | |
| 51 | #[derive(Deserialize)] |
| 52 | struct HookRow { |
| 53 | id: String, |
| 54 | scope: String, |
| 55 | workspace: String, |
| 56 | #[serde(default)] |
| 57 | repo_id: Option<String>, |
| 58 | repo: Option<String>, |
| 59 | url: String, |
| 60 | events: String, |
| 61 | active: u32, |
| 62 | secret: String, |
| 63 | secret_hint: String, |
| 64 | created_by: String, |
| 65 | created_at: String, |
| 66 | last_status: Option<String>, |
| 67 | last_delivered_at: Option<String>, |
| 68 | } |
| 69 | |
| 70 | impl HookRow { |
| 71 | fn events(&self) -> Vec<String> { |
| 72 | serde_json::from_str(&self.events).unwrap_or_else(|_| vec!["*".to_owned()]) |
| 73 | } |
| 74 | |
| 75 | fn to_hook(&self) -> Hook { |
| 76 | Hook { |
| 77 | id: self.id.clone(), |
| 78 | scope: if self.scope == "repo" { HookScope::Repo } else { HookScope::Workspace }, |
| 79 | workspace: self.workspace.clone(), |
| 80 | repo: self.repo.clone(), |
| 81 | url: self.url.clone(), |
| 82 | events: self.events(), |
| 83 | active: self.active != 0, |
| 84 | secret_hint: self.secret_hint.clone(), |
| 85 | created_by: self.created_by.clone(), |
| 86 | created_at: self.created_at.clone(), |
| 87 | last_status: self.last_status.clone(), |
| 88 | last_delivered_at: self.last_delivered_at.clone(), |
| 89 | } |
| 90 | } |
| 91 | } |
| 92 | |
| 93 | #[derive(Deserialize)] |
| 94 | struct DeliveryRow { |
| 95 | id: String, |
| 96 | hook_id: String, |
| 97 | event_id: String, |
| 98 | event: String, |
| 99 | payload: String, |
| 100 | status: String, |
| 101 | attempts: u32, |
| 102 | response_status: Option<u16>, |
| 103 | response_body: Option<String>, |
| 104 | error: Option<String>, |
| 105 | duration_ms: Option<u32>, |
| 106 | created_at: String, |
| 107 | delivered_at: Option<String>, |
| 108 | next_attempt_at: Option<String>, |
| 109 | } |
| 110 | |
| 111 | impl From<DeliveryRow> for HookDelivery { |
| 112 | fn from(row: DeliveryRow) -> Self { |
| 113 | HookDelivery { |
| 114 | id: row.id, |
| 115 | hook_id: row.hook_id, |
| 116 | event_id: row.event_id, |
| 117 | event: row.event, |
| 118 | status: row.status, |
| 119 | attempts: row.attempts, |
| 120 | response_status: row.response_status, |
| 121 | response_body: row.response_body, |
| 122 | error: row.error, |
| 123 | duration_ms: row.duration_ms, |
| 124 | payload: row.payload, |
| 125 | created_at: row.created_at, |
| 126 | delivered_at: row.delivered_at, |
| 127 | next_attempt_at: row.next_attempt_at, |
| 128 | } |
| 129 | } |
| 130 | } |
| 131 | |
| 132 | #[derive(Deserialize)] |
| 133 | struct NameRow { |
| 134 | namespace: String, |
| 135 | name: String, |
| 136 | } |
| 137 | |
| 138 | /// What one attempt to send came to. |
| 139 | struct Attempt { |
| 140 | status: Option<u16>, |
| 141 | body: Option<String>, |
| 142 | error: Option<String>, |
| 143 | duration_ms: u32, |
| 144 | } |
| 145 | |
| 146 | impl Attempt { |
| 147 | fn delivered(&self) -> bool { |
| 148 | self.status.is_some_and(|status| (200..300).contains(&status)) |
| 149 | } |
| 150 | } |
| 151 | |
| 152 | fn optional(value: Option<&str>) -> JsValue { |
| 153 | value.map_or(JsValue::NULL, JsValue::from) |
| 154 | } |
| 155 | |
| 156 | fn fail<T>(code: FailureCode, message: impl Into<String>) -> Outcome<T> { |
| 157 | Outcome::fail(code, message) |
| 158 | } |
| 159 | |
| 160 | /// The workspace itself, as the one asking: it can see its own |
| 161 | /// repositories, private ones too, and no one else's. |
| 162 | fn workspace_viewer(slug: &str) -> Viewer { |
| 163 | Some(User { |
| 164 | id: String::new(), |
| 165 | username: slug.to_owned(), |
| 166 | kind: PrincipalKind::Workspace, |
| 167 | verified: true, |
| 168 | workspaces: vec![Membership::member(slug.to_owned())], |
| 169 | ..User::default() |
| 170 | }) |
| 171 | } |
| 172 | |
| 173 | struct Webhooks { |
| 174 | db: D1Database, |
| 175 | sealer: Option<Sealer>, |
| 176 | repos: Fetcher, |
| 177 | identity: Fetcher, |
| 178 | } |
| 179 | |
| 180 | impl Webhooks { |
| 181 | fn new(env: &Env) -> Result<Self> { |
| 182 | Ok(Webhooks { |
| 183 | db: env.d1("DB")?, |
| 184 | sealer: env.secret("WEBHOOKS_KEY").ok().and_then(|key| Sealer::new(&key.to_string())), |
| 185 | repos: env.service("REPOS")?, |
| 186 | identity: env.service("IDENTITY")?, |
| 187 | }) |
| 188 | } |
| 189 | |
| 190 | // --- Who may do what -------------------------------------------------------- |
| 191 | |
| 192 | /// The repository a repository's webhooks are for, if it is the |
| 193 | /// workspace's and the viewer can see it. |
| 194 | async fn repository(&self, owner: &HookOwner, viewer: &Viewer) -> Result<Option<Repo>> { |
| 195 | let Some(path) = &owner.repo else { |
| 196 | return Ok(None); |
| 197 | }; |
| 198 | let found: Outcome<Repo> = g1t_kit::call( |
| 199 | &self.repos, |
| 200 | "get", |
| 201 | &GetArgs { |
| 202 | path: path.clone(), |
| 203 | viewer: viewer.clone(), |
| 204 | }, |
| 205 | ) |
| 206 | .await?; |
| 207 | Ok(found |
| 208 | .into_result() |
| 209 | .ok() |
| 210 | .filter(|repo| repo.namespace == owner.workspace && repo.fork_of.is_none())) |
| 211 | } |
| 212 | |
| 213 | /// Whether `viewer` may see (or, `managing`, change) `owner`'s webhooks, |
| 214 | /// and the repository when they are a repository's. A repository's |
| 215 | /// need the Admin role on it, to see as to change, since they carry |
| 216 | /// its events out; the workspace's are its members' to see and its |
| 217 | /// owners' to change. |
| 218 | async fn allowed(&self, viewer: &Viewer, owner: &HookOwner, managing: bool) -> Result<Outcome<Option<Repo>>> { |
| 219 | let Some(user) = viewer.as_ref() else { |
| 220 | return Ok(fail(FailureCode::Forbidden, "Sign in to see webhooks.")); |
| 221 | }; |
| 222 | if owner.repo.is_some() { |
| 223 | let Some(repo) = self.repository(owner, viewer).await? else { |
| 224 | return Ok(fail(FailureCode::NotFound, "There is no such repository in this workspace.")); |
| 225 | }; |
| 226 | if !access::can(viewer.as_ref(), &repo, Capability::ManageIntegrations) { |
| 227 | return Ok(fail( |
| 228 | FailureCode::Forbidden, |
| 229 | access::needs(Capability::ManageIntegrations, &format!("{}/{}", repo.namespace, repo.name)), |
| 230 | )); |
| 231 | } |
| 232 | return Ok(Outcome::Ok(Some(repo))); |
| 233 | } |
| 234 | if !user.is_member(&owner.workspace) { |
| 235 | return Ok(fail(FailureCode::Forbidden, format!("Only members of {} can see its webhooks.", owner.workspace))); |
| 236 | } |
| 237 | if managing && user.role_in(&owner.workspace) != Some(Role::Owner) { |
| 238 | return Ok(fail(FailureCode::Forbidden, "Only an owner can manage a workspace's own webhooks.")); |
| 239 | } |
| 240 | Ok(Outcome::Ok(None)) |
| 241 | } |
| 242 | |
| 243 | fn owner(mut owner: HookOwner) -> HookOwner { |
| 244 | owner.workspace = owner.workspace.to_lowercase(); |
| 245 | if let Some(repo) = &mut owner.repo { |
| 246 | repo.namespace = repo.namespace.to_lowercase(); |
| 247 | } |
| 248 | owner |
| 249 | } |
| 250 | |
| 251 | /// The webhook, if it belongs to `owner`. |
| 252 | async fn hook_of(&self, owner: &HookOwner, id: &str) -> Result<Option<HookRow>> { |
| 253 | let row = self |
| 254 | .db |
| 255 | .prepare("SELECT * FROM hooks WHERE id = ? AND workspace = ?") |
| 256 | .bind(&[id.into(), owner.workspace.as_str().into()])? |
| 257 | .first::<HookRow>(None) |
| 258 | .await?; |
| 259 | let repo = owner.repo.as_ref().map(|path| format!("{}/{}", path.namespace, path.name)); |
| 260 | Ok(row.filter(|row| match &repo { |
| 261 | Some(repo) => row.scope == "repo" && row.repo.as_deref().is_some_and(|r| r.eq_ignore_ascii_case(repo)), |
| 262 | None => row.scope == "workspace", |
| 263 | })) |
| 264 | } |
| 265 | |
| 266 | // --- Managing webhooks ------------------------------------------------------ |
| 267 | |
| 268 | async fn list(&self, a: ListArgs) -> Result<Outcome<Vec<Hook>>> { |
| 269 | let owner = Self::owner(a.owner); |
| 270 | if let Outcome::Fail(refused) = self.allowed(&a.viewer, &owner, false).await? { |
| 271 | return Ok(Outcome::Fail(refused)); |
| 272 | } |
| 273 | let rows = match &owner.repo { |
| 274 | Some(path) => self |
| 275 | .db |
| 276 | .prepare("SELECT * FROM hooks WHERE scope = 'repo' AND workspace = ? AND lower(repo) = lower(?) ORDER BY id") |
| 277 | .bind(&[owner.workspace.as_str().into(), format!("{}/{}", path.namespace, path.name).into()])?, |
| 278 | None => self |
| 279 | .db |
| 280 | .prepare("SELECT * FROM hooks WHERE scope = 'workspace' AND workspace = ? ORDER BY id") |
| 281 | .bind(&[owner.workspace.as_str().into()])?, |
| 282 | } |
| 283 | .all() |
| 284 | .await? |
| 285 | .results::<HookRow>()?; |
| 286 | Ok(Outcome::Ok(rows.iter().map(HookRow::to_hook).collect())) |
| 287 | } |
| 288 | |
| 289 | async fn create(&self, a: CreateArgs) -> Result<Outcome<CreatedHook>> { |
| 290 | let owner = Self::owner(a.owner); |
| 291 | let repo = match self.allowed(&Some(a.actor.clone()), &owner, true).await? { |
| 292 | Outcome::Ok(repo) => repo, |
| 293 | Outcome::Fail(refused) => return Ok(Outcome::Fail(refused)), |
| 294 | }; |
| 295 | let Some(sealer) = &self.sealer else { |
| 296 | return Ok(fail(FailureCode::Conflict, "Webhooks are not set up on this g1t: it has no key to keep secrets with.")); |
| 297 | }; |
| 298 | let url = a.url.trim().to_owned(); |
| 299 | if let Err(problem) = deliver::check_url(&url) { |
| 300 | return Ok(fail(FailureCode::Invalid, problem)); |
| 301 | } |
| 302 | let events = match deliver::tidy_events(&a.events) { |
| 303 | Ok(events) => events, |
| 304 | Err(problem) => return Ok(fail(FailureCode::Invalid, problem)), |
| 305 | }; |
| 306 | let given = a.secret.map(|secret| secret.trim().to_owned()).filter(|secret| !secret.is_empty()); |
| 307 | let made = given.is_none(); |
| 308 | let secret = given.unwrap_or_else(|| format!("whsec_{}", g1t_secrets::random_hex(24))); |
| 309 | let now = now_ms(); |
| 310 | let id = new_id("hk", now); |
| 311 | self.db |
| 312 | .prepare( |
| 313 | "INSERT INTO hooks (id, scope, workspace, repo_id, repo, url, events, secret, secret_hint, created_by, created_at) |
| 314 | VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", |
| 315 | ) |
| 316 | .bind(&[ |
| 317 | id.as_str().into(), |
| 318 | if repo.is_some() { "repo" } else { "workspace" }.into(), |
| 319 | owner.workspace.as_str().into(), |
| 320 | optional(repo.as_ref().map(|repo| repo.id.as_str())), |
| 321 | optional(repo.as_ref().map(|repo| format!("{}/{}", repo.namespace, repo.name)).as_deref()), |
| 322 | url.as_str().into(), |
| 323 | serde_json::to_string(&events)?.into(), |
| 324 | sealer.seal(&secret, &id).into(), |
| 325 | g1t_secrets::hint(&secret).into(), |
| 326 | a.actor.username.as_str().into(), |
| 327 | rfc3339(now).into(), |
| 328 | ])? |
| 329 | .run() |
| 330 | .await?; |
| 331 | let Some(row) = self.hook_of(&owner, &id).await? else { |
| 332 | return Ok(fail(FailureCode::NotFound, "The webhook was not saved.")); |
| 333 | }; |
| 334 | // Tells the receiver it is wired up, and shows whether it answers. |
| 335 | self.send_ping(&row).await?; |
| 336 | let row = self.hook_of(&owner, &id).await?.unwrap_or(row); |
| 337 | Ok(Outcome::Ok(CreatedHook { |
| 338 | hook: row.to_hook(), |
| 339 | secret: made.then_some(secret), |
| 340 | })) |
| 341 | } |
| 342 | |
| 343 | async fn update(&self, a: UpdateArgs) -> Result<Outcome<Hook>> { |
| 344 | let owner = Self::owner(a.owner); |
| 345 | if let Outcome::Fail(refused) = self.allowed(&Some(a.actor.clone()), &owner, true).await? { |
| 346 | return Ok(Outcome::Fail(refused)); |
| 347 | } |
| 348 | let Some(row) = self.hook_of(&owner, &a.id).await? else { |
| 349 | return Ok(fail(FailureCode::NotFound, "No such webhook.")); |
| 350 | }; |
| 351 | let url = a.url.map(|url| url.trim().to_owned()).unwrap_or(row.url.clone()); |
| 352 | if let Err(problem) = deliver::check_url(&url) { |
| 353 | return Ok(fail(FailureCode::Invalid, problem)); |
| 354 | } |
| 355 | let events = match a.events { |
| 356 | Some(events) => match deliver::tidy_events(&events) { |
| 357 | Ok(events) => events, |
| 358 | Err(problem) => return Ok(fail(FailureCode::Invalid, problem)), |
| 359 | }, |
| 360 | None => row.events(), |
| 361 | }; |
| 362 | let active = a.active.unwrap_or(row.active != 0); |
| 363 | self.db |
| 364 | .prepare("UPDATE hooks SET url = ?, events = ?, active = ? WHERE id = ?") |
| 365 | .bind(&[url.into(), serde_json::to_string(&events)?.into(), (active as u32).into(), row.id.as_str().into()])? |
| 366 | .run() |
| 367 | .await?; |
| 368 | Ok(match self.hook_of(&owner, &row.id).await? { |
| 369 | Some(row) => Outcome::Ok(row.to_hook()), |
| 370 | None => fail(FailureCode::NotFound, "No such webhook."), |
| 371 | }) |
| 372 | } |
| 373 | |
| 374 | async fn delete(&self, a: HookArgs) -> Result<Outcome<bool>> { |
| 375 | let owner = Self::owner(a.owner); |
| 376 | if let Outcome::Fail(refused) = self.allowed(&Some(a.actor.clone()), &owner, true).await? { |
| 377 | return Ok(Outcome::Fail(refused)); |
| 378 | } |
| 379 | let Some(row) = self.hook_of(&owner, &a.id).await? else { |
| 380 | return Ok(fail(FailureCode::NotFound, "No such webhook.")); |
| 381 | }; |
| 382 | self.db |
| 383 | .batch(vec![ |
| 384 | self.db.prepare("DELETE FROM hooks WHERE id = ?").bind(&[row.id.as_str().into()])?, |
| 385 | self.db.prepare("DELETE FROM deliveries WHERE hook_id = ?").bind(&[row.id.as_str().into()])?, |
| 386 | ]) |
| 387 | .await?; |
| 388 | Ok(Outcome::Ok(true)) |
| 389 | } |
| 390 | |
| 391 | async fn ping(&self, a: HookArgs) -> Result<Outcome<HookDelivery>> { |
| 392 | let owner = Self::owner(a.owner); |
| 393 | if let Outcome::Fail(refused) = self.allowed(&Some(a.actor.clone()), &owner, true).await? { |
| 394 | return Ok(Outcome::Fail(refused)); |
| 395 | } |
| 396 | let Some(row) = self.hook_of(&owner, &a.id).await? else { |
| 397 | return Ok(fail(FailureCode::NotFound, "No such webhook.")); |
| 398 | }; |
| 399 | let id = self.send_ping(&row).await?; |
| 400 | Ok(self.delivery(&id).await?.map_or_else(|| fail(FailureCode::NotFound, "The ping was not recorded."), Outcome::Ok)) |
| 401 | } |
| 402 | |
| 403 | async fn deliveries(&self, a: DeliveriesArgs) -> Result<Outcome<Vec<HookDelivery>>> { |
| 404 | let owner = Self::owner(a.owner); |
| 405 | if let Outcome::Fail(refused) = self.allowed(&a.viewer, &owner, false).await? { |
| 406 | return Ok(Outcome::Fail(refused)); |
| 407 | } |
| 408 | if self.hook_of(&owner, &a.id).await?.is_none() { |
| 409 | return Ok(fail(FailureCode::NotFound, "No such webhook.")); |
| 410 | } |
| 411 | let rows = self |
| 412 | .db |
| 413 | .prepare("SELECT * FROM deliveries WHERE hook_id = ? ORDER BY id DESC LIMIT ?") |
| 414 | .bind(&[a.id.as_str().into(), DELIVERIES_SHOWN.into()])? |
| 415 | .all() |
| 416 | .await? |
| 417 | .results::<DeliveryRow>()?; |
| 418 | Ok(Outcome::Ok(rows.into_iter().map(HookDelivery::from).collect())) |
| 419 | } |
| 420 | |
| 421 | async fn redeliver(&self, a: RedeliverArgs) -> Result<Outcome<HookDelivery>> { |
| 422 | let owner = Self::owner(a.owner); |
| 423 | if let Outcome::Fail(refused) = self.allowed(&Some(a.actor.clone()), &owner, true).await? { |
| 424 | return Ok(Outcome::Fail(refused)); |
| 425 | } |
| 426 | let Some(original) = self.delivery(&a.delivery_id).await? else { |
| 427 | return Ok(fail(FailureCode::NotFound, "No such delivery.")); |
| 428 | }; |
| 429 | let Some(row) = self.hook_of(&owner, &original.hook_id).await? else { |
| 430 | return Ok(fail(FailureCode::NotFound, "No such delivery.")); |
| 431 | }; |
| 432 | // A new delivery of the same payload: its own attempts and log. |
| 433 | let id = self.enqueue(&row, "", &original.event, &original.payload).await?; |
| 434 | if let Some(id) = &id { |
| 435 | self.attempt(&row, id).await?; |
| 436 | } |
| 437 | Ok(match id { |
| 438 | Some(id) => self.delivery(&id).await?.map_or_else(|| fail(FailureCode::NotFound, "No such delivery."), Outcome::Ok), |
| 439 | None => fail(FailureCode::Conflict, "That delivery could not be made again."), |
| 440 | }) |
| 441 | } |
| 442 | |
| 443 | async fn delivery(&self, id: &str) -> Result<Option<HookDelivery>> { |
| 444 | Ok(self |
| 445 | .db |
| 446 | .prepare("SELECT * FROM deliveries WHERE id = ?") |
| 447 | .bind(&[id.into()])? |
| 448 | .first::<DeliveryRow>(None) |
| 449 | .await? |
| 450 | .map(HookDelivery::from)) |
| 451 | } |
| 452 | |
| 453 | // --- Delivering ------------------------------------------------------------- |
| 454 | |
| 455 | /// Records a delivery to make. `None` when this event was already |
| 456 | /// delivered to this webhook. |
| 457 | async fn enqueue(&self, hook: &HookRow, event_id: &str, event: &str, payload: &str) -> Result<Option<String>> { |
| 458 | let now = now_ms(); |
| 459 | let id = new_id("dlv", now); |
| 460 | let inserted = self |
| 461 | .db |
| 462 | .prepare( |
| 463 | "INSERT OR IGNORE INTO deliveries (id, hook_id, event_id, event, payload, status, created_at, next_attempt_at) |
| 464 | VALUES (?, ?, ?, ?, ?, 'pending', ?, ?) RETURNING id", |
| 465 | ) |
| 466 | .bind(&[ |
| 467 | id.as_str().into(), |
| 468 | hook.id.as_str().into(), |
| 469 | event_id.into(), |
| 470 | event.into(), |
| 471 | payload.into(), |
| 472 | rfc3339(now).into(), |
| 473 | rfc3339(now).into(), |
| 474 | ])? |
| 475 | .first::<Value>(None) |
| 476 | .await?; |
| 477 | Ok(inserted.map(|_| id)) |
| 478 | } |
| 479 | |
| 480 | async fn send_ping(&self, hook: &HookRow) -> Result<String> { |
| 481 | let body = deliver::ping(&hook.id, &hook.url, &hook.events(), &rfc3339(now_ms())).to_string(); |
| 482 | let id = self.enqueue(hook, "", "ping", &body).await?.unwrap_or_default(); |
| 483 | self.attempt(hook, &id).await?; |
| 484 | Ok(id) |
| 485 | } |
| 486 | |
| 487 | /// Sends one delivery once, and records how it went. |
| 488 | async fn attempt(&self, hook: &HookRow, delivery_id: &str) -> Result<()> { |
| 489 | let Some(delivery) = self.delivery(delivery_id).await? else { |
| 490 | return Ok(()); |
| 491 | }; |
| 492 | let Some(secret) = self.sealer.as_ref().and_then(|sealer| sealer.open(&hook.secret, &hook.id)) else { |
| 493 | return Ok(()); |
| 494 | }; |
| 495 | let attempt = send(&hook.url, &hook.id, &delivery, &secret).await; |
| 496 | let attempts = delivery.attempts + 1; |
| 497 | let now = now_ms(); |
| 498 | let (status, next) = if attempt.delivered() { |
| 499 | ("delivered", None) |
| 500 | } else { |
| 501 | match deliver::retry_after(attempts) { |
| 502 | Some(wait) => ("pending", Some(rfc3339(now + wait * 1000))), |
| 503 | None => ("failed", None), |
| 504 | } |
| 505 | }; |
| 506 | self.db |
| 507 | .batch(vec![ |
| 508 | self.db |
| 509 | .prepare( |
| 510 | "UPDATE deliveries SET status = ?, attempts = ?, response_status = ?, response_body = ?, error = ?, |
| 511 | duration_ms = ?, delivered_at = ?, next_attempt_at = ? WHERE id = ?", |
| 512 | ) |
| 513 | .bind(&[ |
| 514 | status.into(), |
| 515 | attempts.into(), |
| 516 | attempt.status.map_or(JsValue::NULL, |status| status.into()), |
| 517 | optional(attempt.body.as_deref()), |
| 518 | optional(attempt.error.as_deref()), |
| 519 | attempt.duration_ms.into(), |
| 520 | optional(attempt.delivered().then(|| rfc3339(now)).as_deref()), |
| 521 | optional(next.as_deref()), |
| 522 | delivery_id.into(), |
| 523 | ])?, |
| 524 | self.db |
| 525 | .prepare("UPDATE hooks SET last_status = ?, last_delivered_at = ? WHERE id = ?") |
| 526 | .bind(&[status.into(), rfc3339(now).into(), hook.id.as_str().into()])?, |
| 527 | ]) |
| 528 | .await?; |
| 529 | Ok(()) |
| 530 | } |
| 531 | |
| 532 | /// Which workspace a repository is in, and its name. |
| 533 | async fn repo_name(&self, repo_id: &str, workspaces: &[String]) -> Result<Option<NameRow>> { |
| 534 | if let Some(known) = self |
| 535 | .db |
| 536 | .prepare("SELECT namespace, name FROM repo_names WHERE repo_id = ?") |
| 537 | .bind(&[repo_id.into()])? |
| 538 | .first::<NameRow>(None) |
| 539 | .await? |
| 540 | { |
| 541 | return Ok(Some(known)); |
| 542 | } |
| 543 | // Asked as each workspace with webhooks in turn: each sees only its |
| 544 | // own private repositories. |
| 545 | for workspace in workspaces { |
| 546 | let found: Outcome<Repo> = g1t_kit::call( |
| 547 | &self.repos, |
| 548 | "get_by_id", |
| 549 | &GetByIdArgs { |
| 550 | id: repo_id.to_owned(), |
| 551 | viewer: workspace_viewer(workspace), |
| 552 | }, |
| 553 | ) |
| 554 | .await?; |
| 555 | if let Outcome::Ok(repo) = found |
| 556 | && repo.fork_of.is_none() |
| 557 | { |
| 558 | self.remember(repo_id, &repo.namespace, &repo.name).await?; |
| 559 | return Ok(Some(NameRow { |
| 560 | namespace: repo.namespace, |
| 561 | name: repo.name, |
| 562 | })); |
| 563 | } |
| 564 | } |
| 565 | Ok(None) |
| 566 | } |
| 567 | |
| 568 | async fn remember(&self, repo_id: &str, namespace: &str, name: &str) -> Result<()> { |
| 569 | self.db |
| 570 | .prepare("INSERT OR REPLACE INTO repo_names (repo_id, namespace, name) VALUES (?, ?, ?)") |
| 571 | .bind(&[repo_id.into(), namespace.into(), name.into()])? |
| 572 | .run() |
| 573 | .await?; |
| 574 | Ok(()) |
| 575 | } |
| 576 | |
| 577 | /// An event from the bus, delivered to every webhook that wants it. |
| 578 | async fn on_event(&self, event: &Event) -> Result<()> { |
| 579 | if event.kind == "repo.created" |
| 580 | && let (Some(id), Some(namespace), Some(name)) = |
| 581 | (event.data["repoId"].as_str(), event.data["namespace"].as_str(), event.data["name"].as_str()) |
| 582 | { |
| 583 | self.remember(id, namespace, name).await?; |
| 584 | } |
| 585 | if let Some((workspace, repo_ids)) = deliver::item_scoped(event) { |
| 586 | return self.on_item_event(event, &workspace, &repo_ids).await; |
| 587 | } |
| 588 | let Some(repo_id) = event.repo_id.as_deref().or_else(|| event.data["repoId"].as_str()) else { |
| 589 | // About no repository: a workspace's own package, for its |
| 590 | // workspace's webhooks. |
| 591 | if let Some(workspace) = deliver::workspace_scoped(event) { |
| 592 | let hooks = self |
| 593 | .db |
| 594 | .prepare("SELECT * FROM hooks WHERE active = 1 AND scope = 'workspace' AND workspace = ?") |
| 595 | .bind(&[workspace.as_str().into()])? |
| 596 | .all() |
| 597 | .await? |
| 598 | .results::<HookRow>()?; |
| 599 | self.send_all(event, hooks, None).await?; |
| 600 | } |
| 601 | return Ok(()); |
| 602 | }; |
| 603 | let mut hooks = self |
| 604 | .db |
| 605 | .prepare("SELECT * FROM hooks WHERE active = 1 AND scope = 'repo' AND repo_id = ?") |
| 606 | .bind(&[repo_id.into()])? |
| 607 | .all() |
| 608 | .await? |
| 609 | .results::<HookRow>()?; |
| 610 | let workspaces: Vec<String> = self |
| 611 | .db |
| 612 | .prepare("SELECT DISTINCT workspace FROM hooks WHERE active = 1 AND scope = 'workspace'") |
| 613 | .all() |
| 614 | .await? |
| 615 | .results::<Value>()? |
| 616 | .into_iter() |
| 617 | .filter_map(|row| row["workspace"].as_str().map(str::to_owned)) |
| 618 | .collect(); |
| 619 | let name = if hooks.is_empty() && workspaces.is_empty() { |
| 620 | None |
| 621 | } else { |
| 622 | // A deleted repository looks missing to repos; its own events |
| 623 | // name it. |
| 624 | self.repo_name(repo_id, &workspaces) |
| 625 | .await? |
| 626 | .or_else(|| deliver::named_in(event).map(|(namespace, name)| NameRow { namespace, name })) |
| 627 | }; |
| 628 | if let Some(name) = &name { |
| 629 | hooks.extend( |
| 630 | self.db |
| 631 | .prepare("SELECT * FROM hooks WHERE active = 1 AND scope = 'workspace' AND workspace = ?") |
| 632 | .bind(&[name.namespace.as_str().into()])? |
| 633 | .all() |
| 634 | .await? |
| 635 | .results::<HookRow>()?, |
| 636 | ); |
| 637 | } |
| 638 | self.send_all(event, hooks, Some((repo_id, name.as_ref()))).await |
| 639 | } |
| 640 | |
| 641 | /// An event about a work item: to its workspace's webhooks, and to |
| 642 | /// those of every repository it is linked to. Each repository's |
| 643 | /// webhooks are told of their own repository; the workspace's, of |
| 644 | /// where it was filed, when it was filed in one. |
| 645 | async fn on_item_event(&self, event: &Event, workspace: &str, repo_ids: &[String]) -> Result<()> { |
| 646 | let hooks = self |
| 647 | .db |
| 648 | .prepare( |
| 649 | "SELECT * FROM hooks WHERE active = 1 |
| 650 | AND ((scope = 'workspace' AND workspace = ?1) OR (scope = 'repo' AND repo_id IN (SELECT value FROM json_each(?2))))", |
| 651 | ) |
| 652 | .bind(&[workspace.into(), serde_json::to_string(repo_ids)?.into()])? |
| 653 | .all() |
| 654 | .await? |
| 655 | .results::<HookRow>()?; |
| 656 | if hooks.is_empty() { |
| 657 | return Ok(()); |
| 658 | } |
| 659 | let origin = event.repo_id.as_deref().filter(|id| repo_ids.iter().any(|linked| linked == id)); |
| 660 | let name = match origin { |
| 661 | Some(id) => self.repo_name(id, &[workspace.to_owned()]).await?, |
| 662 | None => None, |
| 663 | }; |
| 664 | self.send_all(event, hooks, origin.map(|id| (id, name.as_ref()))).await |
| 665 | } |
| 666 | |
| 667 | /// Sends `event` to each of `hooks` that wants it. `repo` is the |
| 668 | /// repository it is about, and its name when known; a repository's |
| 669 | /// webhook is told of its own. |
| 670 | async fn send_all(&self, event: &Event, hooks: Vec<HookRow>, repo: Option<(&str, Option<&NameRow>)>) -> Result<()> { |
| 671 | let wanted: Vec<&HookRow> = hooks.iter().filter(|hook| deliver::wants(&hook.events(), &event.kind)).collect(); |
| 672 | if wanted.is_empty() { |
| 673 | return Ok(()); |
| 674 | } |
| 675 | // Who caused it, by name: people and workspaces, or g1t's agent. |
| 676 | let actor_name = match &event.actor { |
| 677 | Some(id) => { |
| 678 | let names: std::collections::HashMap<String, String> = |
| 679 | g1t_kit::call(&self.identity, "usernames", &UsernamesArgs { ids: vec![id.clone()] }).await?; |
| 680 | names.get(id).cloned().or_else(|| (id == AGENT_ID).then(|| AGENT_NAME.to_owned())) |
| 681 | } |
| 682 | None => None, |
| 683 | }; |
| 684 | for hook in wanted { |
| 685 | let own = hook.repo_id.as_deref().filter(|_| hook.scope == "repo").map(|id| (id, None)); |
| 686 | let repo = own.or(repo); |
| 687 | let full_name = repo.map(|(_, name)| { |
| 688 | hook.repo |
| 689 | .clone() |
| 690 | .or_else(|| name.map(|name| format!("{}/{}", name.namespace, name.name))) |
| 691 | .unwrap_or_default() |
| 692 | }); |
| 693 | let named = repo.zip(full_name.as_deref()).map(|((id, _), full)| (id, full)); |
| 694 | let payload = deliver::payload(event, &hook.workspace, named, actor_name.as_deref()).to_string(); |
| 695 | if let Some(id) = self.enqueue(hook, &event.id, &event.kind, &payload).await? { |
| 696 | self.attempt(hook, &id).await?; |
| 697 | } |
| 698 | } |
| 699 | Ok(()) |
| 700 | } |
| 701 | |
| 702 | /// Tries again what is due. |
| 703 | async fn sweep(&self) -> Result<()> { |
| 704 | let now = now_ms(); |
| 705 | let due = self |
| 706 | .db |
| 707 | .prepare("SELECT * FROM deliveries WHERE status = 'pending' AND next_attempt_at <= ? ORDER BY next_attempt_at LIMIT ?") |
| 708 | .bind(&[rfc3339(now).into(), SWEEP.into()])? |
| 709 | .all() |
| 710 | .await? |
| 711 | .results::<DeliveryRow>()?; |
| 712 | for delivery in due { |
| 713 | let hook = self |
| 714 | .db |
| 715 | .prepare("SELECT * FROM hooks WHERE id = ?") |
| 716 | .bind(&[delivery.hook_id.as_str().into()])? |
| 717 | .first::<HookRow>(None) |
| 718 | .await?; |
| 719 | match hook { |
| 720 | Some(hook) if hook.active != 0 => self.attempt(&hook, &delivery.id).await?, |
| 721 | // A webhook turned off or removed takes its retries with it. |
| 722 | _ => { |
| 723 | self.db |
| 724 | .prepare("UPDATE deliveries SET status = 'failed', next_attempt_at = NULL WHERE id = ?") |
| 725 | .bind(&[delivery.id.as_str().into()])? |
| 726 | .run() |
| 727 | .await?; |
| 728 | } |
| 729 | } |
| 730 | } |
| 731 | Ok(()) |
| 732 | } |
| 733 | |
| 734 | /// Forgets deliveries older than [`KEPT_DAYS`], oldest first, a batch |
| 735 | /// at a time (`deliveries_by_time`, migration 0002). What one purge |
| 736 | /// leaves, the next hour's takes. Returns how many it deleted. |
| 737 | async fn purge(&self) -> Result<u32> { |
| 738 | let before = rfc3339(now_ms().saturating_sub(KEPT_DAYS * 24 * 60 * 60 * 1000)); |
| 739 | let mut deleted = 0; |
| 740 | for _ in 0..PURGE_BATCHES { |
| 741 | let result = self |
| 742 | .db |
| 743 | .prepare(PURGE_SQL) |
| 744 | .bind(&[before.as_str().into(), PURGE_BATCH.into()])? |
| 745 | .run() |
| 746 | .await?; |
| 747 | let changed = result.meta()?.and_then(|meta| meta.changes).unwrap_or(0) as u32; |
| 748 | deleted += changed; |
| 749 | if changed < PURGE_BATCH { |
| 750 | break; |
| 751 | } |
| 752 | } |
| 753 | Ok(deleted) |
| 754 | } |
| 755 | } |
| 756 | |
| 757 | /// One HTTPS POST of a delivery's payload, signed, given ten seconds. |
| 758 | async fn send(url: &str, hook_id: &str, delivery: &HookDelivery, secret: &str) -> Attempt { |
| 759 | let started = now_ms(); |
| 760 | let elapsed = || (now_ms() - started) as u32; |
| 761 | let request = (|| -> Result<Request> { |
| 762 | let headers = Headers::new(); |
| 763 | headers.set("content-type", "application/json")?; |
| 764 | headers.set("user-agent", "g1t-webhooks/1 (+https://docs.g1t.sh/guides/webhooks/)")?; |
| 765 | headers.set("x-g1t-event", &delivery.event)?; |
| 766 | headers.set("x-g1t-delivery", &delivery.id)?; |
| 767 | headers.set("x-g1t-hook", hook_id)?; |
| 768 | headers.set("x-g1t-signature-256", &deliver::signature(secret, &delivery.payload))?; |
| 769 | let mut init = RequestInit::new(); |
| 770 | init.with_method(Method::Post).with_headers(headers).with_body(Some(delivery.payload.clone().into())); |
| 771 | Request::new_with_init(url, &init) |
| 772 | })(); |
| 773 | let request = match request { |
| 774 | Ok(request) => request, |
| 775 | Err(error) => { |
| 776 | return Attempt { |
| 777 | status: None, |
| 778 | body: None, |
| 779 | error: Some(format!("The request could not be made: {error}")), |
| 780 | duration_ms: 0, |
| 781 | }; |
| 782 | } |
| 783 | }; |
| 784 | let fetcher = Fetch::Request(request); |
| 785 | let fetch = Box::pin(fetcher.send()); |
| 786 | let timeout = Box::pin(Delay::from(TIMEOUT)); |
| 787 | match select(fetch, timeout).await { |
| 788 | Either::Left((Ok(mut response), _)) => { |
| 789 | let status = response.status_code(); |
| 790 | let body: String = response.text().await.unwrap_or_default().chars().take(RESPONSE_KEPT).collect(); |
| 791 | Attempt { |
| 792 | status: Some(status), |
| 793 | body: Some(body), |
| 794 | error: None, |
| 795 | duration_ms: elapsed(), |
| 796 | } |
| 797 | } |
| 798 | Either::Left((Err(error), _)) => Attempt { |
| 799 | status: None, |
| 800 | body: None, |
| 801 | error: Some(format!("The receiver could not be reached: {error}")), |
| 802 | duration_ms: elapsed(), |
| 803 | }, |
| 804 | Either::Right(_) => Attempt { |
| 805 | status: None, |
| 806 | body: None, |
| 807 | error: Some(format!("The receiver did not answer within {} seconds.", TIMEOUT.as_secs())), |
| 808 | duration_ms: elapsed(), |
| 809 | }, |
| 810 | } |
| 811 | } |
| 812 | |
| 813 | #[event(fetch)] |
| 814 | async fn fetch(mut request: Request, env: Env, _ctx: Context) -> Result<Response> { |
| 815 | let Some(method) = rpc_method(&request) else { |
| 816 | return Response::error("Not found", 404); |
| 817 | }; |
| 818 | let body: Value = request.json().await?; |
| 819 | let service = Webhooks::new(&env)?; |
| 820 | match method.as_str() { |
| 821 | "list" => reply(&service.list(args(body)?).await?), |
| 822 | "create" => reply(&service.create(args(body)?).await?), |
| 823 | "update" => reply(&service.update(args(body)?).await?), |
| 824 | "delete" => reply(&service.delete(args(body)?).await?), |
| 825 | "ping" => reply(&service.ping(args(body)?).await?), |
| 826 | "deliveries" => reply(&service.deliveries(args(body)?).await?), |
| 827 | "redeliver" => reply(&service.redeliver(args(body)?).await?), |
| 828 | _ => Response::error("Unknown method", 404), |
| 829 | } |
| 830 | } |
| 831 | |
| 832 | /// Events from the bus, on this service's own queue. |
| 833 | #[event(queue)] |
| 834 | async fn queue(batch: MessageBatch<Event>, env: Env, _ctx: Context) -> Result<()> { |
| 835 | let service = Webhooks::new(&env)?; |
| 836 | for message in batch.messages()? { |
| 837 | // A workspace renamed: its rows move to the slug it has now. |
| 838 | if g1t_kit::rename::on_event(&env, &env.d1("DB")?, message.body(), rename::STATEMENTS).await? { |
| 839 | message.ack(); |
| 840 | continue; |
| 841 | } |
| 842 | // A repository renamed or transferred: its rows follow its new path, |
| 843 | // and then the event is delivered like any other, naming it there. |
| 844 | if g1t_kit::transfer::on_event(&env, &env.d1("DB")?, message.body(), rename::TRANSFERRED).await? { |
| 845 | service.on_event(message.body()).await?; |
| 846 | message.ack(); |
| 847 | continue; |
| 848 | } |
| 849 | // A repository purged: its last event is delivered, then its own |
| 850 | // webhooks go. |
| 851 | if message.body().kind == "repo.purged" { |
| 852 | service.on_event(message.body()).await?; |
| 853 | g1t_kit::lifecycle::on_purged(&env.d1("DB")?, message.body(), rename::PURGED).await?; |
| 854 | message.ack(); |
| 855 | continue; |
| 856 | } |
| 857 | // A workspace deleted: what it kept for itself goes. |
| 858 | if g1t_kit::deleted::on_event(&env.d1("DB")?, message.body(), rename::DELETED).await? { |
| 859 | message.ack(); |
| 860 | continue; |
| 861 | } |
| 862 | service.on_event(message.body()).await?; |
| 863 | message.ack(); |
| 864 | } |
| 865 | Ok(()) |
| 866 | } |
| 867 | |
| 868 | /// One batch of old deliveries, by time. |
| 869 | const PURGE_SQL: &str = "DELETE FROM deliveries WHERE rowid IN (SELECT rowid FROM deliveries WHERE created_at < ? ORDER BY created_at LIMIT ?)"; |
| 870 | |
| 871 | /// Every minute: retries that are due. Once an hour, at [`PURGE_CRON`]: |
| 872 | /// deliveries old enough to forget. |
| 873 | #[event(scheduled)] |
| 874 | async fn scheduled(event: ScheduledEvent, env: Env, _ctx: ScheduleContext) { |
| 875 | let service = match Webhooks::new(&env) { |
| 876 | Ok(service) => service, |
| 877 | Err(error) => { |
| 878 | worker::console_error!("webhooks: could not start: {error}"); |
| 879 | return; |
| 880 | } |
| 881 | }; |
| 882 | if event.cron() == PURGE_CRON { |
| 883 | match service.purge().await { |
| 884 | Ok(0) => {} |
| 885 | Ok(deleted) => worker::console_log!("webhooks: forgot {deleted} old deliveries"), |
| 886 | Err(error) => worker::console_error!("webhooks: the purge failed: {error}"), |
| 887 | } |
| 888 | return; |
| 889 | } |
| 890 | if let Err(error) = service.sweep().await { |
| 891 | worker::console_error!("webhooks: the sweep failed: {error}"); |
| 892 | } |
| 893 | } |
| 894 | |
| 895 | #[cfg(test)] |
| 896 | mod tests { |
| 897 | use super::*; |
| 898 | |
| 899 | #[test] |
| 900 | fn the_purge_has_a_cron_of_its_own() { |
| 901 | let wrangler = include_str!("../wrangler.jsonc"); |
| 902 | assert!(wrangler.contains(&format!("\"{PURGE_CRON}\"")), "{PURGE_CRON} is not among the crons"); |
| 903 | assert_ne!(PURGE_CRON, "* * * * *"); |
| 904 | } |
| 905 | |
| 906 | #[test] |
| 907 | fn the_purge_deletes_in_batches_by_time() { |
| 908 | assert!(PURGE_SQL.contains("WHERE created_at < ? ORDER BY created_at LIMIT ?")); |
| 909 | assert!(PURGE_BATCH * PURGE_BATCHES >= 10_000); |
| 910 | } |
| 911 | } |
| 912 | |