Skip to content
Sign up

flagon-io/g1t

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