steid
Sign in

@jamesgill /

steid

steid/src/application/webhook.rs
Rust22.1 KBCode·Blame·History·Raw
1//! Registering webhooks, and delivering a push to them.
2//!
3//! A webhook is the one place Steid makes an outbound connection to an address a person
4//! typed. Everything defensive about that lives either in
5//! [`WebhookUrl`](crate::domain::WebhookUrl) — which refuses the address at the moment
6//! it is saved, when there is somebody to tell — or in the sending adapter, which
7//! re-checks what the name resolved to.
8//!
9//! **Nothing here survives a restart.** Retries live in the task that started them, so a
10//! process that stops between attempts loses the remaining ones; the delivery rows
11//! written so far stay, so what happened is still readable. A durable queue is the
12//! answer when deliveries have to be reliable, and it is a different feature.
13
14use std::time::{Duration, SystemTime};
15
16use futures_util::future::join_all;
17
18use crate::domain::{
19 Actor, DomainError, ObjectId, OrgName, PrivateTargets, PushEvent, PushEventId, RepoName,
20 Webhook, WebhookDelivery, WebhookDeliveryId, WebhookId, WebhookSecret, WebhookUrl, is_zero,
21 repository::{
22 MembershipRepository, OrgRepository, PushEventRepository, RepoRepository, UserRepository,
23 WebhookDeliveryRepository, WebhookRepository,
24 },
25};
26
27use super::{
28 access::{Stores, changeable_repo},
29 error::{Error, Result},
30 identity::describe_identity,
31 paging::{Paged, clamp, offset, page_count},
32 port::{Clock, GitQuery, WebhookOutcome, WebhookRequest, WebhookSender},
33};
34
35/// The payload's own version, carried in every body.
36///
37/// Steid's shape rather than an imitation of GitHub's: a receiver that thinks it is
38/// talking to GitHub and finds three quarters of the fields missing is worse served
39/// than one told plainly what this is. The number is what lets the shape change later
40/// without every receiver having to guess.
41pub const PAYLOAD_VERSION: u32 = 1;
42
43/// How many commits a payload carries.
44///
45/// A body is a thing a receiver parses in a request handler, and a push of ten thousand
46/// commits must not become a ten-megabyte POST. The `before`/`after` pair is in the
47/// payload precisely so a receiver that wants the rest can ask git for it.
48pub const MAX_PAYLOAD_COMMITS: usize = 20;
49
50/// How long to wait before each retry.
51///
52/// One initial attempt plus these, so four sends at most and a delivery row for each.
53/// The spread is deliberate: 5 seconds catches a receiver that was restarting, 30 a
54/// deploy, and 2 minutes an outage short enough to be worth waiting out. Anything longer
55/// belongs to a durable queue rather than to a task holding a `Vec` in memory.
56pub const RETRY_BACKOFF: [Duration; 3] = [
57 Duration::from_secs(5),
58 Duration::from_secs(30),
59 Duration::from_secs(120),
60];
61
62/// How many attempts one page of the deliveries list shows.
63pub const DELIVERY_LIMIT: usize = 50;
64
65/// A hook as it appears in a listing.
66///
67/// Carries no secret. The secret is shown once, at creation, and a list has nothing to
68/// do with it.
69#[derive(Debug, Clone, PartialEq, Eq)]
70pub struct WebhookSummary {
71 pub id: WebhookId,
72 pub url: WebhookUrl,
73 pub active: bool,
74 pub created_at: SystemTime,
75 /// The most recent attempt, which is what the status dot reads. `None` for a hook
76 /// nothing has been pushed to yet — a real state, and a different one from failing.
77 pub last: Option<WebhookDelivery>,
78}
79
80/// A hook and its recent attempts.
81#[derive(Debug, Clone, PartialEq, Eq)]
82pub struct WebhookDetail {
83 pub id: WebhookId,
84 pub url: WebhookUrl,
85 pub active: bool,
86 pub created_at: SystemTime,
87 /// One page of the hook's attempts, newest first.
88 pub deliveries: Paged<WebhookDelivery>,
89}
90
91/// A hook and the one and only chance to read its secret.
92#[derive(Debug)]
93pub struct CreatedWebhook {
94 pub webhook: Webhook,
95 pub secret: WebhookSecret,
96}
97
98/// Registers a hook on a repository.
99///
100/// Owner only, through the same predicate the settings page uses — a member's read
101/// access is not permission to point someone's pushes at a server they chose.
102///
103/// The secret is minted here rather than accepted from the form. A person choosing it
104/// would choose a weak one, and the value has no purpose outside signing.
105///
106/// `targets` is [`PrivateTargets::Refused`] on any instance that has not deliberately
107/// opted out — the caller reads it from configuration rather than this deciding, because
108/// the domain must not know that an environment variable exists.
109#[allow(clippy::too_many_arguments)]
110pub async fn create_webhook(
111 actor: &Actor,
112 handle: &OrgName,
113 name: &RepoName,
114 url: &str,
115 now: SystemTime,
116 targets: PrivateTargets,
117 stores: &Stores<'_, impl OrgRepository, impl MembershipRepository, impl RepoRepository>,
118 webhooks: &impl WebhookRepository,
119) -> Result<CreatedWebhook> {
120 let repo = changeable_repo(actor, handle, name, stores).await?;
121
122 let secret = WebhookSecret::generate();
123 let webhook = Webhook::new(
124 WebhookId::generate(),
125 repo.id,
126 url,
127 secret.clone(),
128 now,
129 targets,
130 )?;
131
132 webhooks.save(&webhook).await?;
133
134 Ok(CreatedWebhook { webhook, secret })
135}
136
137/// Every hook on a repository, with the outcome of its most recent attempt.
138///
139/// One extra query per hook. A repository has a handful of hooks, and the alternative —
140/// a join expressed in both repository implementations — buys nothing at that size.
141pub async fn list_webhooks(
142 actor: &Actor,
143 handle: &OrgName,
144 name: &RepoName,
145 stores: &Stores<'_, impl OrgRepository, impl MembershipRepository, impl RepoRepository>,
146 webhooks: &impl WebhookRepository,
147 deliveries: &impl WebhookDeliveryRepository,
148) -> Result<Vec<WebhookSummary>> {
149 let repo = changeable_repo(actor, handle, name, stores).await?;
150
151 let mut listed = Vec::new();
152
153 for webhook in webhooks.list_by_repo(&repo.id).await? {
154 let last = deliveries.latest_by_webhook(&webhook.id).await?;
155
156 listed.push(WebhookSummary {
157 id: webhook.id,
158 url: webhook.url,
159 active: webhook.active,
160 created_at: webhook.created_at,
161 last,
162 });
163 }
164
165 Ok(listed)
166}
167
168/// One hook and its recent attempts.
169///
170/// A hook belonging to a *different* repository answers `NotFound`, for the reason a
171/// token belonging to someone else does: that an id exists elsewhere is not a fact this
172/// caller should be able to learn by trying ids.
173#[allow(clippy::too_many_arguments)]
174pub async fn view_webhook(
175 actor: &Actor,
176 handle: &OrgName,
177 name: &RepoName,
178 id: &WebhookId,
179 page: usize,
180 stores: &Stores<'_, impl OrgRepository, impl MembershipRepository, impl RepoRepository>,
181 webhooks: &impl WebhookRepository,
182 deliveries: &impl WebhookDeliveryRepository,
183) -> Result<WebhookDetail> {
184 let repo = changeable_repo(actor, handle, name, stores).await?;
185
186 let webhook = webhooks
187 .find(id)
188 .await?
189 .filter(|webhook| webhook.repo_id == repo.id)
190 .ok_or_else(missing)?;
191
192 let total = deliveries.count_by_webhook(&webhook.id).await?;
193 let page = clamp(page, page_count(total, DELIVERY_LIMIT));
194 let recorded = deliveries
195 .list_by_webhook(&webhook.id, offset(page, DELIVERY_LIMIT), DELIVERY_LIMIT)
196 .await?;
197
198 Ok(WebhookDetail {
199 deliveries: Paged::of(recorded, total, page, DELIVERY_LIMIT),
200 id: webhook.id,
201 url: webhook.url,
202 active: webhook.active,
203 created_at: webhook.created_at,
204 })
205}
206
207/// Removes a hook, and its delivery history with it.
208pub async fn delete_webhook(
209 actor: &Actor,
210 handle: &OrgName,
211 name: &RepoName,
212 id: &WebhookId,
213 stores: &Stores<'_, impl OrgRepository, impl MembershipRepository, impl RepoRepository>,
214 webhooks: &impl WebhookRepository,
215) -> Result<()> {
216 let repo = changeable_repo(actor, handle, name, stores).await?;
217
218 let webhook = webhooks
219 .find(id)
220 .await?
221 .filter(|webhook| webhook.repo_id == repo.id)
222 .ok_or_else(missing)?;
223
224 webhooks.delete(&webhook.id).await?;
225
226 Ok(())
227}
228
229/// Turns a hook on or off, and answers with the hook as it now stands.
230///
231/// Idempotent: the caller passes the state it wants rather than "the other one", so a
232/// double submit — a reload, a second click — lands on the same answer instead of
233/// flipping back. The page reads the current state when it renders the button, which is
234/// what makes that the natural shape here.
235///
236/// A hook that is off keeps its deliveries and its secret. That is the whole reason for
237/// having an off switch rather than only a delete: the delivery log is usually the
238/// evidence of *why* it was turned off.
239pub async fn set_webhook_active(
240 actor: &Actor,
241 handle: &OrgName,
242 name: &RepoName,
243 id: &WebhookId,
244 active: bool,
245 stores: &Stores<'_, impl OrgRepository, impl MembershipRepository, impl RepoRepository>,
246 webhooks: &impl WebhookRepository,
247) -> Result<Webhook> {
248 let repo = changeable_repo(actor, handle, name, stores).await?;
249
250 let mut webhook = webhooks
251 .find(id)
252 .await?
253 .filter(|webhook| webhook.repo_id == repo.id)
254 .ok_or_else(missing)?;
255
256 webhook.active = active;
257 webhooks.save(&webhook).await?;
258
259 Ok(webhook)
260}
261
262/// Sends one recorded event to one hook again, as a fresh attempt.
263///
264/// **The payload is rebuilt, not replayed from a column.** A delivery row keeps what
265/// came back, not what went out — storing every body would put a copy of the repository's
266/// commit list in the database once per attempt — so the body is assembled from the same
267/// [`PushEvent`] by the same [`push_payload`] the original used. In practice that is the
268/// same bytes; where it can differ is the commit list, which is read from git now and
269/// would notice history having been rewritten under it. That is the honest answer for a
270/// redelivery: it says what the repository holds today.
271///
272/// One attempt and no retry schedule. A retry is what happens to a delivery nobody asked
273/// about; this one has somebody watching the page, and they can press the button again.
274///
275/// The attempt number continues the event's own sequence rather than restarting at 1, so
276/// the log reads as "the fifth time we tried this push" rather than as a second first
277/// try. It is taken from the attempts already recorded against this hook — the same page
278/// of them the deliveries page shows, which is also where the button was pressed.
279///
280/// **An inactive hook can still be redelivered to.** Turning a hook off stops Steid
281/// calling it on its own; a person pressing Redeliver is asking for exactly this one
282/// call, which is usually how you check whether the endpoint is fixed before turning it
283/// back on.
284#[allow(clippy::too_many_arguments)]
285pub async fn redeliver(
286 actor: &Actor,
287 handle: &OrgName,
288 name: &RepoName,
289 hook_id: &WebhookId,
290 delivery_id: &WebhookDeliveryId,
291 // The origin this instance is reached on, e.g. `https://steid.example` — the same
292 // value a push derives its payload's repository URL from.
293 origin: &str,
294 stores: &Stores<'_, impl OrgRepository, impl MembershipRepository, impl RepoRepository>,
295 users: &impl UserRepository,
296 webhooks: &impl WebhookRepository,
297 deliveries: &impl WebhookDeliveryRepository,
298 events: &impl PushEventRepository,
299 git: &impl GitQuery,
300 sender: &impl WebhookSender,
301 clock: &impl Clock,
302) -> Result<WebhookDelivery> {
303 let repo = changeable_repo(actor, handle, name, stores).await?;
304
305 let hook = webhooks
306 .find(hook_id)
307 .await?
308 .filter(|hook| hook.repo_id == repo.id)
309 .ok_or_else(missing)?;
310
311 // One read serves three purposes: finding the attempt the button names, learning
312 // which event it belongs to, and counting how many attempts that event has had. The
313 // button only ever appears on a row this page rendered, so the row being outside the
314 // window is the same "no such attempt" as it never having existed.
315 // The whole history, not a page of it: the button is on a row the reader is looking
316 // at, which may be on any page, and the attempt number below counts every attempt
317 // the *event* has had — a number the old 50-row window got wrong as soon as an event
318 // fell out of it. See `plans/current.md` for what this costs a long-lived hook.
319 let recorded = deliveries.list_by_webhook(&hook.id, 0, usize::MAX).await?;
320
321 let event_id: PushEventId = recorded
322 .iter()
323 .find(|delivery| &delivery.id == delivery_id)
324 .map(|delivery| delivery.event_id.clone())
325 .ok_or(DomainError::NotFound {
326 entity: "webhook delivery",
327 })?;
328
329 let event = events
330 .find(&event_id)
331 .await?
332 .filter(|event| event.repo_id == repo.id)
333 .ok_or(DomainError::NotFound {
334 entity: "push event",
335 })?;
336
337 let attempt = recorded
338 .iter()
339 .filter(|delivery| delivery.event_id == event_id)
340 .map(|delivery| delivery.attempt)
341 .max()
342 .unwrap_or(0)
343 + 1;
344
345 let pusher = match &event.pusher {
346 Some(user_id) => describe_identity(&Actor::User(user_id.clone()), users, stores.orgs)
347 .await?
348 .map(|identity| identity.handle.to_string()),
349 None => None,
350 };
351
352 let context = DeliveryContext {
353 handle: handle.clone(),
354 name: name.clone(),
355 repo_url: format!("{origin}/{handle}/repos/{name}"),
356 pusher,
357 };
358
359 let commits = commits_for(&event, &context, git).await;
360 let body = push_payload(&event, &context, &commits);
361
362 Ok(deliver_one(
363 &hook,
364 &event,
365 &body,
366 deliveries,
367 sender,
368 clock,
369 &[],
370 attempt,
371 )
372 .await)
373}
374
375/// What a payload needs to know about the world outside the event.
376#[derive(Debug, Clone)]
377pub struct DeliveryContext {
378 pub handle: OrgName,
379 pub name: RepoName,
380 /// The repository's own page, absolute — a receiver has no way to build one.
381 /// Derived from the request the push arrived on, so an instance never has to be
382 /// told its own address.
383 pub repo_url: String,
384 /// The handle of whoever pushed. `None` only if their user record has gone between
385 /// the push and the delivery.
386 pub pusher: Option<String>,
387}
388
389/// Builds the JSON body for one event.
390///
391/// Steid's own shape, versioned — see [`PAYLOAD_VERSION`]. Timestamps are Unix seconds
392/// rather than a formatted date: Steid has no date formatter, and a receiver parsing an
393/// integer cannot be caught out by a timezone.
394pub fn push_payload(
395 event: &PushEvent,
396 context: &DeliveryContext,
397 commits: &[crate::domain::CommitSummary],
398) -> Vec<u8> {
399 let body = serde_json::json!({
400 "version": PAYLOAD_VERSION,
401 "event": "push",
402 "repository": {
403 "handle": context.handle.as_str(),
404 "name": context.name.as_str(),
405 "url": context.repo_url,
406 },
407 "ref": event.ref_name.as_str(),
408 "before": event.before.as_str(),
409 "after": event.after.as_str(),
410 "commits": commits
411 .iter()
412 .map(|commit| {
413 serde_json::json!({
414 "id": commit.id.as_str(),
415 "summary": commit.summary,
416 "author": commit.author_name,
417 "timestamp": unix(commit.committed_at),
418 })
419 })
420 .collect::<Vec<_>>(),
421 "pusher": { "handle": context.pusher },
422 });
423
424 // `to_vec`, not `to_string`: the signature is over the bytes actually sent, and
425 // re-encoding a string to sign it is how the two drift apart.
426 serde_json::to_vec(&body).unwrap_or_else(|_| b"{}".to_vec())
427}
428
429/// The headers every delivery carries.
430///
431/// `X-Steid-Delivery` is the **push event's** id, not the attempt's, so all four
432/// attempts at one event carry the same value — deduplicating on it is the whole reason
433/// a receiver would read it, and an id that changed per retry could not be used for
434/// that.
435fn delivery_headers(
436 event: &PushEvent,
437 secret: &WebhookSecret,
438 body: &[u8],
439) -> Vec<(String, String)> {
440 vec![
441 ("Content-Type".to_owned(), "application/json".to_owned()),
442 ("X-Steid-Event".to_owned(), "push".to_owned()),
443 ("X-Steid-Delivery".to_owned(), event.id.to_string()),
444 ("X-Steid-Signature-256".to_owned(), secret.sign(body)),
445 ]
446}
447
448/// Delivers every event of one push to every active hook on the repository.
449///
450/// Takes no `Actor`: its only input is a list of [`PushEvent`]s, which nothing but
451/// [`record_push`](super::push::record_push) produces and which that function will not
452/// produce for anyone but an owner. Authorization is settled upstream, once.
453///
454/// **Hooks run concurrently, events run in order.** A dead endpoint holds its own task
455/// for two and a half minutes of backoff, and must not hold anybody else's; but two
456/// pushes to one hook arriving out of order would be a receiver's bug to work around
457/// forever.
458///
459/// `backoff` is a parameter rather than a constant so the tests do not sleep. Production
460/// passes [`RETRY_BACKOFF`].
461#[allow(clippy::too_many_arguments)]
462pub async fn deliver_push(
463 events: &[PushEvent],
464 context: &DeliveryContext,
465 git: &impl GitQuery,
466 webhooks: &impl WebhookRepository,
467 deliveries: &impl WebhookDeliveryRepository,
468 sender: &impl WebhookSender,
469 clock: &impl Clock,
470 backoff: &[Duration],
471) -> Result<()> {
472 let Some(first) = events.first() else {
473 return Ok(());
474 };
475
476 let hooks: Vec<Webhook> = webhooks
477 .list_by_repo(&first.repo_id)
478 .await?
479 .into_iter()
480 .filter(|hook| hook.active)
481 .collect();
482
483 if hooks.is_empty() {
484 return Ok(());
485 }
486
487 // Built once for all hooks: the body is identical, and only the signature differs.
488 // Asking git for the same commits once per hook would multiply the forks by the
489 // number of endpoints somebody happened to register.
490 let mut bodies = Vec::with_capacity(events.len());
491
492 for event in events {
493 let commits = commits_for(event, context, git).await;
494 bodies.push((event, push_payload(event, context, &commits)));
495 }
496
497 join_all(hooks.iter().map(|hook| async {
498 for (event, body) in &bodies {
499 deliver_one(hook, event, body, deliveries, sender, clock, backoff, 1).await;
500 }
501 }))
502 .await;
503
504 Ok(())
505}
506
507/// The commits a payload lists, newest first.
508///
509/// A deletion has none. A creation is diffed against nothing, which is every commit the
510/// new ref reaches — capped like any other, because a first push of a large repository
511/// would otherwise put its whole history in a POST.
512///
513/// A read that fails costs the payload its commit list and nothing else: the ref, the
514/// two ids and the repository are the parts a receiver cannot reconstruct, and refusing
515/// to deliver at all because git was slow would be the wrong trade.
516async fn commits_for(
517 event: &PushEvent,
518 context: &DeliveryContext,
519 git: &impl GitQuery,
520) -> Vec<crate::domain::CommitSummary> {
521 if event.is_delete() {
522 return Vec::new();
523 }
524
525 let base: Option<&ObjectId> = (!is_zero(&event.before)).then_some(&event.before);
526
527 match git
528 .log_between(
529 &context.handle,
530 &context.name,
531 base,
532 &event.after,
533 0,
534 MAX_PAYLOAD_COMMITS,
535 )
536 .await
537 {
538 Ok(commits) => commits,
539 Err(error) => {
540 eprintln!(
541 "steid: could not list the commits of a push to {}/{}: {error}",
542 context.handle, context.name
543 );
544 Vec::new()
545 }
546 }
547}
548
549/// Sends one event to one hook, retrying on failure, recording every attempt.
550///
551/// Never returns an error. A delivery that could not be made is a row saying so, and a
552/// row that could not be written is a log line — neither is something the push that
553/// caused it should ever hear about. The last attempt is handed back for the one caller
554/// that has somebody waiting on the answer — [`redeliver`] — and ignored by the push
555/// path, which has nobody to tell.
556///
557/// `first` is the number the first attempt is recorded under. A push starts at 1; a
558/// redelivery continues the event's own sequence, so its row does not claim to be a
559/// first try.
560#[allow(clippy::too_many_arguments)]
561async fn deliver_one(
562 hook: &Webhook,
563 event: &PushEvent,
564 body: &[u8],
565 deliveries: &impl WebhookDeliveryRepository,
566 sender: &impl WebhookSender,
567 clock: &impl Clock,
568 backoff: &[Duration],
569 first: u32,
570) -> WebhookDelivery {
571 let mut recorded = None;
572
573 for attempt in 0..=backoff.len() {
574 if attempt > 0 {
575 tokio::time::sleep(backoff[attempt - 1]).await;
576 }
577
578 let outcome: WebhookOutcome = sender
579 .send(WebhookRequest {
580 url: hook.url.as_str().to_owned(),
581 headers: delivery_headers(event, &hook.secret, body),
582 body: body.to_vec(),
583 })
584 .await;
585
586 let delivery = WebhookDelivery::record(
587 WebhookDeliveryId::generate(),
588 hook.id.clone(),
589 event.id.clone(),
590 first + attempt as u32,
591 outcome.status,
592 &outcome.body,
593 outcome.duration.as_millis().min(u128::from(u64::MAX)) as u64,
594 // Asked per attempt, not once for the delivery: attempts are minutes apart
595 // and a single timestamp would have the deliveries page claim four of them
596 // in one instant. This is the case the [`Clock`] port exists for.
597 clock.now(),
598 );
599
600 let succeeded = delivery.ok;
601
602 if let Err(error) = deliveries.save(&delivery).await {
603 eprintln!("steid: a webhook delivery could not be recorded: {error}");
604 }
605
606 recorded = Some(delivery);
607
608 if succeeded {
609 break;
610 }
611 }
612
613 // The loop body runs at least once — `0..=backoff.len()` is never empty — so there
614 // is always an attempt to report.
615 recorded.expect("a delivery always makes at least one attempt")
616}
617
618fn unix(time: SystemTime) -> i64 {
619 time.duration_since(SystemTime::UNIX_EPOCH)
620 .map(|elapsed| elapsed.as_secs() as i64)
621 .unwrap_or(0)
622}
623
624fn missing() -> Error {
625 DomainError::NotFound { entity: "webhook" }.into()
626}
627
628#[cfg(test)]
629#[path = "webhook_tests.rs"]
630mod tests;