Skip to content
Sign up

flagon-io/g1t

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