Skip to content
Sign up

flagon-io/g1t

public
0
548 linesCodeBlameRaw
1//! The events service: the bus every state change in g1t is published on,
2//! and its durable log.
3//!
4//! Publishing puts events on a queue and returns. The queue consumer writes
5//! them to the log and passes each batch on to each subscriber's own
6//! queue, so a slow or failing subscriber holds up nobody else. Each
7//! subscriber is sent only the types it acts on
8//! (`g1t_contracts::subscribers`), in batches the queues take (fanout.rs).
9//!
10//! It keeps two more things beside the log: the audit log (audit.rs) and
11//! each person's inbox (inbox.rs), written as events arrive, with who
12//! follows what (subscriptions.rs).
13//!
14//! Other services reach it over `POST /rpc/<method>`; see
15//! `g1t_contracts::events`, `audit` and `inbox` for the methods and their
16//! arguments. `activity_digest` (digest.rs) counts a span of the log for
17//! Home. Security alerts that are not urgent wait for each person's
18//! security digest (security_digest.rs), sent by a cron of its own.
19//!
20//! Repository pages open in someone's browser refresh themselves: each
21//! batch's events that such a page shows are passed to the notify service
22//! (live.rs), which tells those watching the repository.
23
24mod audit;
25mod digest;
26mod fanout;
27mod inbox;
28mod live;
29mod security_digest;
30mod subscriptions;
31
32use g1t_contracts::events::{Event, ListArgs, PublishArgs};
33use g1t_contracts::new_id;
34use g1t_contracts::time::rfc3339;
35use g1t_kit::{args, js, now_ms, reply, rpc_method};
36use serde::Deserialize;
37use std::collections::{HashMap, HashSet};
38use worker::js_sys::{Array, Object};
39use worker::wasm_bindgen::{JsCast, JsValue};
40use worker::{Context, D1Database, Env, MessageBatch, Request, Response, Result, event};
41
42const DEFAULT_PAGE: u32 = 50;
43const MAX_PAGE: u32 = 200;
44/// Statuses and check runs reported on commits: delivered to webhooks, and
45/// left out of a timeline unless asked for by type.
46const REPORTING: [&str; 7] = [
47 "status.created",
48 "check_run.created",
49 "check_run.completed",
50 "check_run.rerequested",
51 "check_run.requested_action",
52 "check_suite.completed",
53 "check_suite.rerequested",
54];
55/// Every binding whose name starts with this is a queue that receives
56/// events: one per subscribing service, sent the types it routes.
57const SUBSCRIBER_PREFIX: &str = "SUBSCRIBER_";
58/// How long a record of which queues a batch reached is kept for its retries.
59const FANOUT_KEEP_MS: u64 = 24 * 60 * 60 * 1000;
60
61/// An event as the log's table holds it.
62#[derive(Deserialize)]
63pub(crate) struct EventRow {
64 id: String,
65 #[serde(rename = "type")]
66 kind: String,
67 source: String,
68 time: String,
69 repo_id: Option<String>,
70 actor: Option<String>,
71 /// JSON.
72 data: String,
73}
74
75impl From<EventRow> for Event {
76 fn from(row: EventRow) -> Self {
77 Event {
78 id: row.id,
79 kind: row.kind,
80 source: row.source,
81 time: row.time,
82 repo_id: row.repo_id,
83 actor: row.actor,
84 data: serde_json::from_str(&row.data).unwrap_or_default(),
85 }
86 }
87}
88
89fn optional(value: &Option<String>) -> JsValue {
90 value.as_deref().map_or(JsValue::NULL, JsValue::from)
91}
92
93#[derive(Deserialize)]
94struct FanoutRow {
95 event_id: String,
96 /// JSON array of binding names.
97 bindings: String,
98}
99
100/// Sends `events` to a queue binding, one message each, in as many
101/// `sendBatch` calls as the queue's limits need (fanout.rs). Events are
102/// expected to have been fit to a message.
103async fn send(queue: &JsValue, events: &[&Event]) -> Result<()> {
104 for chunk in fanout::chunks(events) {
105 let messages = Array::new();
106 for event in chunk {
107 let message = Object::new();
108 js::set(&message, "body", &js::to_js(event)?);
109 messages.push(&message);
110 }
111 js::call(queue, "sendBatch", &[messages.into()]).await?;
112 }
113 Ok(())
114}
115
116struct Events {
117 db: D1Database,
118 env: Env,
119}
120
121impl Events {
122 /// Assigns each event its id and time and puts it on the bus.
123 async fn publish(&self, a: PublishArgs) -> Result<()> {
124 if a.events.is_empty() {
125 return Ok(());
126 }
127 let now = now_ms();
128 let events: Vec<Event> = a
129 .events
130 .into_iter()
131 .map(|event| {
132 let mut event = Event {
133 id: new_id("evt", now),
134 kind: event.kind,
135 source: event.source,
136 time: rfc3339(now),
137 repo_id: event.repo_id,
138 actor: event.actor,
139 data: event.data,
140 };
141 // Too large for one message: its long text is shortened.
142 fanout::fit(&mut event);
143 event
144 })
145 .collect();
146 let events: Vec<&Event> = events.iter().collect();
147 send(&js::binding(&self.env, "BUS")?, &events).await
148 }
149
150 /// Newest first. Callers must have checked that the viewer may see the
151 /// repository asked about.
152 async fn list(&self, a: ListArgs) -> Result<Vec<Event>> {
153 let (filter, binds) = list_filter(&a);
154 let mut values: Vec<JsValue> = binds
155 .into_iter()
156 .map(|bind| match bind {
157 Bind::Text(text) => JsValue::from(text),
158 Bind::Number(number) => JsValue::from(number),
159 })
160 .collect();
161 values.push(a.limit.unwrap_or(DEFAULT_PAGE).min(MAX_PAGE).into());
162 let rows = self
163 .db
164 .prepare(format!(
165 "SELECT * FROM events {filter} ORDER BY id DESC LIMIT ?"
166 ))
167 .bind(&values)?
168 .all()
169 .await?
170 .results::<EventRow>()?;
171 Ok(rows.into_iter().map(Event::from).collect())
172 }
173
174 /// Writes a batch from the bus to the log, notes who now follows what
175 /// (inbox.rs), hands each subscriber the events it routes, then tells
176 /// the people it concerns.
177 ///
178 /// Everything before the hand-off is safe to repeat. A retried batch
179 /// skips the queues that already have it: when a hand-off fails part
180 /// way, which queues each event reached is written down first.
181 async fn deliver(&self, events: &[Event]) -> Result<()> {
182 let mut statements = Vec::with_capacity(events.len() + 1);
183 for event in events {
184 statements.push(
185 self.db
186 .prepare(
187 // Redelivered batches must not duplicate log rows.
188 "INSERT OR IGNORE INTO events (id, type, source, time, repo_id, actor, data)
189 VALUES (?, ?, ?, ?, ?, ?, ?)",
190 )
191 .bind(&[
192 event.id.as_str().into(),
193 event.kind.as_str().into(),
194 event.source.as_str().into(),
195 event.time.as_str().into(),
196 optional(&event.repo_id),
197 optional(&event.actor),
198 serde_json::to_string(&event.data)?.into(),
199 ])?,
200 );
201 }
202 let ids: Vec<&str> = events.iter().map(|event| event.id.as_str()).collect();
203 // Read in the same round trip: nearly always nothing.
204 statements.push(
205 self.db
206 .prepare("SELECT event_id, bindings FROM fanout_sent WHERE event_id IN (SELECT value FROM json_each(?))")
207 .bind(&[serde_json::to_string(&ids)?.into()])?,
208 );
209 let results = self.db.batch(statements).await?;
210 let mut sent: HashMap<String, HashSet<String>> = HashMap::new();
211 if let Some(rows) = results.last() {
212 for row in rows.results::<FanoutRow>()? {
213 let bindings: Vec<String> = serde_json::from_str(&row.bindings).unwrap_or_default();
214 sent.insert(row.event_id, bindings.into_iter().collect());
215 }
216 }
217 audit::follow_renames(&self.db, events).await?;
218 // Before the hand-off, so its failing does not send the batch to
219 // every queue again.
220 inbox::follow(&self.db, events).await?;
221
222 let bindings: &JsValue = self.env.as_ref();
223 let mut reached: Vec<String> = Vec::new();
224 let mut failed = None;
225 for name in Object::keys(bindings.unchecked_ref::<Object>()).iter() {
226 let Some(name) = name
227 .as_string()
228 .filter(|name| name.starts_with(SUBSCRIBER_PREFIX))
229 else {
230 continue;
231 };
232 let routed: Vec<&Event> = events
233 .iter()
234 .filter(|event| g1t_contracts::subscribers::routed(&name, &event.kind))
235 .filter(|event| !sent.get(&event.id).is_some_and(|reached| reached.contains(&name)))
236 .collect();
237 if routed.is_empty() {
238 continue;
239 }
240 match send(&js::binding(&self.env, &name)?, &routed).await {
241 Ok(()) => reached.push(name),
242 Err(error) => {
243 worker::console_error!("events: passing {} events to {name} failed: {error}", routed.len());
244 failed = Some(error);
245 }
246 }
247 }
248 if let Some(error) = failed {
249 self.note_reached(events, &sent, &reached).await?;
250 return Err(error);
251 }
252 let (work, repos, identity) = (
253 self.env.service("WORK")?,
254 self.env.service("REPOS")?,
255 self.env.service("IDENTITY")?,
256 );
257 // Optional: without it the inbox still fills, nobody is told live.
258 let notify = self.env.service("NOTIFY").ok();
259 // Pages open on the repositories: told first, as nobody waits longer.
260 if let Some(notify) = notify.as_ref() {
261 live::tell(notify, events).await;
262 }
263 let sources = inbox::Sources {
264 work: &work,
265 repos: &repos,
266 identity: &identity,
267 notify: notify.as_ref(),
268 };
269 inbox::deliver(&self.db, &sources, events).await;
270 Ok(())
271 }
272
273 /// Writes down which queues each event of a batch has reached, with
274 /// those it had reached before, for the batch's retry.
275 async fn note_reached(
276 &self,
277 events: &[Event],
278 sent: &HashMap<String, HashSet<String>>,
279 reached: &[String],
280 ) -> Result<()> {
281 let now = rfc3339(now_ms());
282 let mut statements = Vec::with_capacity(events.len());
283 for event in events {
284 let mut bindings: Vec<&str> = reached.iter().map(String::as_str).collect();
285 if let Some(before) = sent.get(&event.id) {
286 bindings.extend(before.iter().map(String::as_str));
287 }
288 bindings.sort_unstable();
289 bindings.dedup();
290 statements.push(
291 self.db
292 .prepare(
293 "INSERT INTO fanout_sent (event_id, bindings, created_at) VALUES (?, ?, ?)
294 ON CONFLICT (event_id) DO UPDATE SET bindings = excluded.bindings",
295 )
296 .bind(&[event.id.as_str().into(), serde_json::to_string(&bindings)?.into(), now.as_str().into()])?,
297 );
298 }
299 self.db.batch(statements).await?;
300 Ok(())
301 }
302}
303
304#[event(fetch)]
305async fn fetch(mut request: Request, env: Env, _ctx: Context) -> Result<Response> {
306 let Some(method) = rpc_method(&request) else {
307 return Response::error("Not found", 404);
308 };
309 let body: serde_json::Value = request.json().await?;
310 let events = Events {
311 db: env.d1("DB")?,
312 env,
313 };
314 match method.as_str() {
315 "publish" => reply(&events.publish(args(body)?).await?),
316 "list" => reply(&events.list(args(body)?).await?),
317 // Home's digest of a span: counted, by who did it (digest.rs).
318 "activity_digest" => reply(&digest::activity_digest(&events.db, args(body)?).await?),
319 "audit_record" => reply(&audit::record(&events.db, args(body)?).await?),
320 "audit_list" => reply(&audit::list(&events.db, args(body)?).await?),
321 "inbox_list" => {
322 let repos = events.env.service("REPOS")?;
323 reply(&inbox::list(&events.db, &repos, args(body)?).await?)
324 }
325 "inbox_counts" => reply(&inbox::counts(&events.db, args(body)?).await?),
326 "inbox_mark" => {
327 let marked: g1t_contracts::inbox::MarkInboxArgs = args(body)?;
328 let username = marked.username.clone();
329 let changed = inbox::mark(&events.db, marked).await?;
330 // Notify: the new count in every open tab, whoever marked (the site, the API, MCP).
331 if changed > 0
332 && let Ok(notify) = events.env.service("NOTIFY")
333 {
334 inbox::tell_inbox_count(&events.db, &notify, &username).await;
335 }
336 reply(&changed)
337 }
338 "inbox_thread" => {
339 let (repos, work) = (events.env.service("REPOS")?, events.env.service("WORK")?);
340 reply(&inbox::thread(&events.db, &repos, &work, args(body)?).await?)
341 }
342 "inbox_subscription" => {
343 let work = events.env.service("WORK")?;
344 reply(&subscriptions::subscription(&events.db, &work, args(body)?).await?)
345 }
346 "inbox_subscribe" => {
347 let work = events.env.service("WORK")?;
348 reply(&subscriptions::subscribe(&events.db, &work, args(body)?).await?)
349 }
350 "inbox_watching" => reply(&subscriptions::watching(&events.db, args(body)?).await?),
351 "inbox_watch" => reply(&subscriptions::watch(&events.db, args(body)?).await?),
352 "inbox_watchers" => reply(&subscriptions::watchers_count(&events.db, args(body)?).await?),
353 "inbox_watched" => reply(&subscriptions::watched(&events.db, args(body)?).await?),
354 "inbox_settings" => reply(&subscriptions::settings(&events.db, args(body)?).await?),
355 "inbox_update_settings" => reply(&subscriptions::update_settings(&events.db, args(body)?).await?),
356 _ => Response::error("Unknown method", 404),
357 }
358}
359
360/// Events from the bus. A batch that fails is retried whole, which is safe
361/// because the log ignores an event it has seen, and queues that already
362/// have it are skipped (`deliver`). After its retries it goes to the
363/// dead-letter queue (wrangler.jsonc).
364#[event(queue)]
365async fn queue(batch: MessageBatch<Event>, env: Env, _ctx: Context) -> Result<()> {
366 let events = Events {
367 db: env.d1("DB")?,
368 env,
369 };
370 let delivered: Vec<Event> = batch
371 .messages()?
372 .into_iter()
373 .map(|message| message.into_body())
374 .collect();
375 events.deliver(&delivered).await?;
376 batch.ack_all();
377 Ok(())
378}
379
380/// Once a day, old audit entries are removed: first everything older than
381/// any workspace keeps (`AUDIT_MAX_DAYS`, 400 days), then each workspace's
382/// entries older than its own plan keeps, as billing says (7 days free, 90
383/// on the plan, or what staff set). Workspaces with nothing older than the
384/// shortest (`AUDIT_MIN_DAYS`, 7) are left alone.
385#[event(scheduled)]
386async fn scheduled(event: worker::ScheduledEvent, env: Env, _ctx: worker::ScheduleContext) {
387 // The morning cron sends the security digests that are due, and nothing else.
388 if event.cron() == security_digest::DIGEST_CRON {
389 let (Ok(db), Ok(identity)) = (env.d1("DB"), env.service("IDENTITY")) else { return };
390 let notify = env.service("NOTIFY").ok();
391 match security_digest::send_due(&db, &identity, notify.as_ref(), now_ms()).await {
392 Ok(sent) if sent > 0 => worker::console_log!("sent {sent} security digests"),
393 Ok(_) => {}
394 Err(error) => worker::console_error!("security digests not sent: {error}"),
395 }
396 return;
397 }
398 let days = |name: &str, default: u32| {
399 env.var(name)
400 .ok()
401 .and_then(|v| v.to_string().trim().parse::<u32>().ok())
402 .filter(|days| *days > 0)
403 .unwrap_or(default)
404 };
405 let max_days = days("AUDIT_MAX_DAYS", audit::DEFAULT_MAX_DAYS);
406 let min_days = days("AUDIT_MIN_DAYS", audit::DEFAULT_MIN_DAYS).min(max_days);
407 let Ok(db) = env.d1("DB") else { return };
408 let now = now_ms();
409 let cutoff = rfc3339(now.saturating_sub(FANOUT_KEEP_MS));
410 let purged = match db.prepare("DELETE FROM fanout_sent WHERE created_at < ?").bind(&[cutoff.into()]) {
411 Ok(statement) => statement.run().await.map(|_| ()),
412 Err(error) => Err(error),
413 };
414 if let Err(error) = purged {
415 worker::console_error!("could not remove old fan-out records: {error}");
416 }
417 match inbox::purge(&db, now).await {
418 Ok(removed) if removed > 0 => worker::console_log!("removed {removed} old inbox items"),
419 Ok(_) => {}
420 Err(error) => worker::console_error!("could not remove old inbox items: {error}"),
421 }
422 match audit::purge(&db, &audit::keep_from(now, max_days), 20).await {
423 Ok(removed) if removed > 0 => {
424 worker::console_log!("removed {removed} audit entries older than {max_days} days")
425 }
426 Ok(_) => {}
427 Err(error) => worker::console_error!("could not remove old audit entries: {error}"),
428 }
429 // Without billing nobody's plan is known, so nothing younger than the
430 // ceiling is removed.
431 let billing = match env.service("BILLING") {
432 Ok(billing) => billing,
433 Err(error) => {
434 worker::console_error!(
435 "audit entries kept past their plan's days: no billing: {error}"
436 );
437 return;
438 }
439 };
440 match audit::purge_by_plan(&db, &billing, now, min_days, max_days).await {
441 Ok(removed) if removed > 0 => {
442 worker::console_log!("removed {removed} audit entries older than their plan keeps")
443 }
444 Ok(_) => {}
445 Err(error) => worker::console_error!("audit entries kept past their plan's days: {error}"),
446 }
447}
448
449/// A value bound to a `?` in [`list_filter`]'s clause.
450#[derive(Debug, PartialEq)]
451enum Bind {
452 Text(String),
453 Number(f64),
454}
455
456/// The `WHERE` clause `list` reads with, and what it binds, in order.
457fn list_filter(a: &ListArgs) -> (String, Vec<Bind>) {
458 let mut conditions = Vec::new();
459 let mut values = Vec::new();
460 if let Some(repo_id) = &a.repo_id {
461 conditions.push("repo_id = ?".to_owned());
462 values.push(Bind::Text(repo_id.clone()));
463 }
464 if !a.types.is_empty() {
465 let marks = vec!["?"; a.types.len()].join(", ");
466 conditions.push(format!("type IN ({marks})"));
467 values.extend(a.types.iter().map(|kind| Bind::Text(kind.clone())));
468 } else {
469 // What CI and integrations report on commits goes to webhooks,
470 // and is read from each commit's checks; a timeline asked for
471 // everything would be little else on a busy repository.
472 let marks = vec!["?"; REPORTING.len()].join(", ");
473 conditions.push(format!("type NOT IN ({marks})"));
474 values.extend(REPORTING.iter().map(|kind| Bind::Text((*kind).to_owned())));
475 }
476 if let Some(actor) = &a.actor {
477 conditions.push("actor = ?".to_owned());
478 values.push(Bind::Text(actor.clone()));
479 }
480 if !a.numbers.is_empty() {
481 // An issue or pull request's own events name it as `number`; a
482 // comment, review or link on it names it as `issue`.
483 let marks = vec!["?"; a.numbers.len()].join(", ");
484 conditions.push(format!(
485 "(json_extract(data, '$.number') IN ({marks}) OR json_extract(data, '$.issue') IN ({marks}))"
486 ));
487 for _ in 0..2 {
488 values.extend(a.numbers.iter().map(|number| Bind::Number(f64::from(*number))));
489 }
490 }
491 if let Some(since) = &a.since {
492 conditions.push("time >= ?".to_owned());
493 values.push(Bind::Text(since.clone()));
494 }
495 if let Some(before) = &a.before {
496 conditions.push("id < ?".to_owned());
497 values.push(Bind::Text(before.clone()));
498 }
499 let filter = if conditions.is_empty() {
500 String::new()
501 } else {
502 format!("WHERE {}", conditions.join(" AND "))
503 };
504 (filter, values)
505}
506
507#[cfg(test)]
508mod tests {
509 use super::*;
510
511 #[test]
512 fn a_plain_list_leaves_out_what_is_reported_on_commits() {
513 let (filter, binds) = list_filter(&ListArgs { repo_id: Some("rep_1".into()), ..ListArgs::default() });
514 assert!(filter.starts_with("WHERE repo_id = ? AND type NOT IN ("));
515 assert_eq!(binds[0], Bind::Text("rep_1".into()));
516 assert_eq!(binds.len(), 1 + REPORTING.len());
517 }
518
519 #[test]
520 fn numbers_match_an_item_or_what_is_said_on_it_since_a_time() {
521 let (filter, binds) = list_filter(&ListArgs {
522 repo_id: Some("rep_1".into()),
523 types: vec!["issue.opened".into()],
524 numbers: vec![4, 9],
525 since: Some("2026-10-01T00:00:00Z".into()),
526 actor: Some("usr_g1t_agent".into()),
527 ..ListArgs::default()
528 });
529 assert_eq!(
530 filter,
531 "WHERE repo_id = ? AND type IN (?) AND actor = ? AND \
532 (json_extract(data, '$.number') IN (?, ?) OR json_extract(data, '$.issue') IN (?, ?)) AND time >= ?"
533 );
534 assert_eq!(
535 binds,
536 vec![
537 Bind::Text("rep_1".into()),
538 Bind::Text("issue.opened".into()),
539 Bind::Text("usr_g1t_agent".into()),
540 Bind::Number(4.0),
541 Bind::Number(9.0),
542 Bind::Number(4.0),
543 Bind::Number(9.0),
544 Bind::Text("2026-10-01T00:00:00Z".into()),
545 ]
546 );
547 }
548}