Skip to content
Sign up

flagon-io/g1t

public
0
526 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.

Stripe webhooks, enterprise invoices, and sudo for both1//! 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
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails5//! 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).
Stripe webhooks, enterprise invoices, and sudo for both9//!
Inbox: the events service tells people what needs them as events arrive10//! It keeps two more things beside the log: the audit log (audit.rs) and
Inbox: threads, reasons, subscriptions and watching11//! each person's inbox (inbox.rs), written as events arrive, with who
12//! follows what (subscriptions.rs).
Inbox: the events service tells people what needs them as events arrive13//!
Stripe webhooks, enterprise invoices, and sudo for both14//! Other services reach it over `POST /rpc/<method>`; see
Inbox: the events service tells people what needs them as events arrive15//! `g1t_contracts::events`, `audit` and `inbox` for the methods and their
Home says what people did as well as what agents did. Since you were last here now has a People column and an Agents column: who pushed how many commits to which projects, pull requests opened, merged and reviewed, issues opened and closed, docs edited, messages sent and deploys that went out, each line a link to where those are listed, with the agents' acceptance (first time, after review, didn't finish) kept as a row of their column; the sentence under the heading sums it up honestly, and says when a part of g1t could not be read, or that the span was quiet. Landed counts merged pull requests, commits pushed straight to a default branch, production deploys that went live, releases and packages, newest first; Running now adds workflow runs. Behind it, the events service answers an activity digest over a span in one round trip from its existing indexes, every push now records how many commits it carried, and chat counts the messages sent in the conversations you can read. The Home guide defines every line, and says how this scales.16//! arguments. `activity_digest` (digest.rs) counts a span of the log for
17//! Home.
Stripe webhooks, enterprise invoices, and sudo for both18
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API19mod audit;
Home says what people did as well as what agents did. Since you were last here now has a People column and an Agents column: who pushed how many commits to which projects, pull requests opened, merged and reviewed, issues opened and closed, docs edited, messages sent and deploys that went out, each line a link to where those are listed, with the agents' acceptance (first time, after review, didn't finish) kept as a row of their column; the sentence under the heading sums it up honestly, and says when a part of g1t could not be read, or that the span was quiet. Landed counts merged pull requests, commits pushed straight to a default branch, production deploys that went live, releases and packages, newest first; Running now adds workflow runs. Behind it, the events service answers an activity digest over a span in one round trip from its existing indexes, every push now records how many commits it carried, and chat counts the messages sent in the conversations you can read. The Home guide defines every line, and says how this scales.20mod digest;
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails21mod fanout;
Inbox: the events service tells people what needs them as events arrive22mod inbox;
Inbox: threads, reasons, subscriptions and watching23mod subscriptions;
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API24
Stripe webhooks, enterprise invoices, and sudo for both25use 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;
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails30use std::collections::{HashMap, HashSet};
Stripe webhooks, enterprise invoices, and sudo for both31use 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;
Merge checks: statuses and check runs on every commit37/// 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];
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails48/// Every binding whose name starts with this is a queue that receives
49/// events: one per subscribing service, sent the types it routes.
Stripe webhooks, enterprise invoices, and sudo for both50const SUBSCRIBER_PREFIX: &str = "SUBSCRIBER_";
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails51/// How long a record of which queues a batch reached is kept for its retries.
52const FANOUT_KEEP_MS: u64 = 24 * 60 * 60 * 1000;
Stripe webhooks, enterprise invoices, and sudo for both53
Home says what people did as well as what agents did. Since you were last here now has a People column and an Agents column: who pushed how many commits to which projects, pull requests opened, merged and reviewed, issues opened and closed, docs edited, messages sent and deploys that went out, each line a link to where those are listed, with the agents' acceptance (first time, after review, didn't finish) kept as a row of their column; the sentence under the heading sums it up honestly, and says when a part of g1t could not be read, or that the span was quiet. Landed counts merged pull requests, commits pushed straight to a default branch, production deploys that went live, releases and packages, newest first; Running now adds workflow runs. Behind it, the events service answers an activity digest over a span in one round trip from its existing indexes, every push now records how many commits it carried, and chat counts the messages sent in the conversations you can read. The Home guide defines every line, and says how this scales.54/// An event as the log's table holds it.
Stripe webhooks, enterprise invoices, and sudo for both55#[derive(Deserialize)]
Home says what people did as well as what agents did. Since you were last here now has a People column and an Agents column: who pushed how many commits to which projects, pull requests opened, merged and reviewed, issues opened and closed, docs edited, messages sent and deploys that went out, each line a link to where those are listed, with the agents' acceptance (first time, after review, didn't finish) kept as a row of their column; the sentence under the heading sums it up honestly, and says when a part of g1t could not be read, or that the span was quiet. Landed counts merged pull requests, commits pushed straight to a default branch, production deploys that went live, releases and packages, newest first; Running now adds workflow runs. Behind it, the events service answers an activity digest over a span in one round trip from its existing indexes, every push now records how many commits it carried, and chat counts the messages sent in the conversations you can read. The Home guide defines every line, and says how this scales.56pub(crate) struct EventRow {
Stripe webhooks, enterprise invoices, and sudo for both57 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
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails86#[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?;
Stripe webhooks, enterprise invoices, and sudo for both105 }
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()
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails124 .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
Stripe webhooks, enterprise invoices, and sudo for both137 })
138 .collect();
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails139 let events: Vec<&Event> = events.iter().collect();
Stripe webhooks, enterprise invoices, and sudo for both140 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>> {
Merge leftovers: plan activity filtered in the query, pushes counted by account, ghost during the deletion window, profile time zones (identity 0039)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();
Stripe webhooks, enterprise invoices, and sudo for both154 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
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails167 /// 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.
Stripe webhooks, enterprise invoices, and sudo for both174 async fn deliver(&self, events: &[Event]) -> Result<()> {
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails175 let mut statements = Vec::with_capacity(events.len() + 1);
Stripe webhooks, enterprise invoices, and sudo for both176 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 }
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails195 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 }
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API210 audit::follow_renames(&self.db, events).await?;
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails211 // Before the hand-off, so its failing does not send the batch to
212 // every queue again.
213 inbox::follow(&self.db, events).await?;
Stripe webhooks, enterprise invoices, and sudo for both214
215 let bindings: &JsValue = self.env.as_ref();
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails216 let mut reached: Vec<String> = Vec::new();
217 let mut failed = None;
Stripe webhooks, enterprise invoices, and sudo for both218 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 };
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails225 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);
Stripe webhooks, enterprise invoices, and sudo for both244 }
Inbox: the events service tells people what needs them as events arrive245 let (work, repos, identity) = (
246 self.env.service("WORK")?,
247 self.env.service("REPOS")?,
248 self.env.service("IDENTITY")?,
249 );
Merge the workspace shell: navigation and phone shell, g1t as orchestrator, agents in roles with audience-checked reads, reactions and custom emoji, live notifications and browser push, the homepage tour (agents 0002, chat 0002)250 // Optional: without it the inbox still fills, nobody is told live.
251 let notify = self.env.service("NOTIFY").ok();
Inbox: the events service tells people what needs them as events arrive252 let sources = inbox::Sources {
253 work: &work,
254 repos: &repos,
255 identity: &identity,
Merge the workspace shell: navigation and phone shell, g1t as orchestrator, agents in roles with audience-checked reads, reactions and custom emoji, live notifications and browser push, the homepage tour (agents 0002, chat 0002)256 notify: notify.as_ref(),
Inbox: the events service tells people what needs them as events arrive257 };
258 inbox::deliver(&self.db, &sources, events).await;
Stripe webhooks, enterprise invoices, and sudo for both259 Ok(())
260 }
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails261
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 }
Stripe webhooks, enterprise invoices, and sudo for both291}
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?),
Home says what people did as well as what agents did. Since you were last here now has a People column and an Agents column: who pushed how many commits to which projects, pull requests opened, merged and reviewed, issues opened and closed, docs edited, messages sent and deploys that went out, each line a link to where those are listed, with the agents' acceptance (first time, after review, didn't finish) kept as a row of their column; the sentence under the heading sums it up honestly, and says when a part of g1t could not be read, or that the span was quiet. Landed counts merged pull requests, commits pushed straight to a default branch, production deploys that went live, releases and packages, newest first; Running now adds workflow runs. Behind it, the events service answers an activity digest over a span in one round trip from its existing indexes, every push now records how many commits it carried, and chat counts the messages sent in the conversations you can read. The Home guide defines every line, and says how this scales.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?),
Agents get guardrails, run credentials, an audit log, a context hub, repository instructions and mentions; security upkeep; snake_case API308 "audit_record" => reply(&audit::record(&events.db, args(body)?).await?),
309 "audit_list" => reply(&audit::list(&events.db, args(body)?).await?),
Inbox: the events service tells people what needs them as events arrive310 "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?),
Merge the workspace shell: navigation and phone shell, g1t as orchestrator, agents in roles with audience-checked reads, reactions and custom emoji, live notifications and browser push, the homepage tour (agents 0002, chat 0002)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 }
Inbox: threads, reasons, subscriptions and watching327 "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?),
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb97341 "inbox_watchers" => reply(&subscriptions::watchers_count(&events.db, args(body)?).await?),
Inbox: threads, reasons, subscriptions and watching342 "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?),
Stripe webhooks, enterprise invoices, and sudo for both345 _ => Response::error("Unknown method", 404),
346 }
347}
348
349/// Events from the bus. A batch that fails is retried whole, which is safe
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails350/// 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).
Stripe webhooks, enterprise invoices, and sudo for both353#[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}
Team plan, an open-source pool, monthly trials and honest metering; the sidebar for everyone; a workspace that stays put368
Audit logs are kept by plan: a week on free, 90 days on the plan, and what staff set for an account in sudo369/// 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.
Team plan, an open-source pool, monthly trials and honest metering; the sidebar for everyone; a workspace that stays put374#[event(scheduled)]
375async fn scheduled(_event: worker::ScheduledEvent, env: Env, _ctx: worker::ScheduleContext) {
Audit logs are kept by plan: a week on free, 90 days on the plan, and what staff set for an account in sudo376 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);
Team plan, an open-source pool, monthly trials and honest metering; the sidebar for everyone; a workspace that stays put385 let Ok(db) = env.d1("DB") else { return };
Audit logs are kept by plan: a week on free, 90 days on the plan, and what staff set for an account in sudo386 let now = now_ms();
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails387 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 }
Inbox: the events service tells people what needs them as events arrive395 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 }
Audit logs are kept by plan: a week on free, 90 days on the plan, and what staff set for an account in sudo400 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 }
Team plan, an open-source pool, monthly trials and honest metering; the sidebar for everyone; a workspace that stays put404 Ok(_) => {}
405 Err(error) => worker::console_error!("could not remove old audit entries: {error}"),
406 }
Audit logs are kept by plan: a week on free, 90 days on the plan, and what staff set for an account in sudo407 // 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 }
Team plan, an open-source pool, monthly trials and honest metering; the sidebar for everyone; a workspace that stays put425}
Merge leftovers: plan activity filtered in the query, pushes counted by account, ghost during the deletion window, profile time zones (identity 0039)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}

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