Skip to content
Sign up

flagon-io/g1t

public
00

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

g1t.sh
gitagentscloudflarerustforgemcp
878 linesCodeBlameRaw

Pick any line to see why it is the way it is: the commit, the pull request and issue it came from, and what the agent was thinking.

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

This file's history is long; its oldest lines are credited to the oldest commit read.