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
548 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.

A last-commits walk answers within 2.5 s with what it has, and keeps only finished answers1//! 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).
A last-commits walk answers within 2.5 s with what it has, and keeps only finished answers9//!
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//!
A last-commits walk answers within 2.5 s with what it has, and keeps only finished answers14//! 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
Merge remote-tracking branch 'origin/claude/cool-newton-7nm8r9' into claude/cool-newton-7nm8r917//! 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.
Merge branch 'worktree-agent-a5c94df323642630e'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.
A last-commits walk answers within 2.5 s with what it has, and keeps only finished answers23
24mod 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.25mod digest;
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails26mod fanout;
Inbox: the events service tells people what needs them as events arrive27mod inbox;
Merge branch 'worktree-agent-a5c94df323642630e'28mod live;
Merge remote-tracking branch 'origin/claude/cool-newton-7nm8r9' into claude/cool-newton-7nm8r929mod security_digest;
Inbox: threads, reasons, subscriptions and watching30mod subscriptions;
A last-commits walk answers within 2.5 s with what it has, and keeps only finished answers31
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;
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails37use std::collections::{HashMap, HashSet};
A last-commits walk answers within 2.5 s with what it has, and keeps only finished answers38use 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;
Merge checks: statuses and check runs on every commit44/// 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];
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails55/// Every binding whose name starts with this is a queue that receives
56/// events: one per subscribing service, sent the types it routes.
A last-commits walk answers within 2.5 s with what it has, and keeps only finished answers57const SUBSCRIBER_PREFIX: &str = "SUBSCRIBER_";
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails58/// How long a record of which queues a batch reached is kept for its retries.
59const FANOUT_KEEP_MS: u64 = 24 * 60 * 60 * 1000;
A last-commits walk answers within 2.5 s with what it has, and keeps only finished answers60
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.61/// An event as the log's table holds it.
A last-commits walk answers within 2.5 s with what it has, and keeps only finished answers62#[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.63pub(crate) struct EventRow {
A last-commits walk answers within 2.5 s with what it has, and keeps only finished answers64 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
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails93#[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?;
A last-commits walk answers within 2.5 s with what it has, and keeps only finished answers112 }
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()
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails131 .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
A last-commits walk answers within 2.5 s with what it has, and keeps only finished answers144 })
145 .collect();
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails146 let events: Vec<&Event> = events.iter().collect();
A last-commits walk answers within 2.5 s with what it has, and keeps only finished answers147 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>> {
Merge leftovers: plan activity filtered in the query, pushes counted by account, ghost during the deletion window, profile time zones (identity 0039)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();
A last-commits walk answers within 2.5 s with what it has, and keeps only finished answers161 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
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails174 /// 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.
A last-commits walk answers within 2.5 s with what it has, and keeps only finished answers181 async fn deliver(&self, events: &[Event]) -> Result<()> {
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails182 let mut statements = Vec::with_capacity(events.len() + 1);
A last-commits walk answers within 2.5 s with what it has, and keeps only finished answers183 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 }
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails202 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 }
A last-commits walk answers within 2.5 s with what it has, and keeps only finished answers217 audit::follow_renames(&self.db, events).await?;
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails218 // Before the hand-off, so its failing does not send the batch to
219 // every queue again.
220 inbox::follow(&self.db, events).await?;
A last-commits walk answers within 2.5 s with what it has, and keeps only finished answers221
222 let bindings: &JsValue = self.env.as_ref();
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails223 let mut reached: Vec<String> = Vec::new();
224 let mut failed = None;
A last-commits walk answers within 2.5 s with what it has, and keeps only finished answers225 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 };
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails232 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);
A last-commits walk answers within 2.5 s with what it has, and keeps only finished answers251 }
Inbox: the events service tells people what needs them as events arrive252 let (work, repos, identity) = (
253 self.env.service("WORK")?,
254 self.env.service("REPOS")?,
255 self.env.service("IDENTITY")?,
256 );
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)257 // Optional: without it the inbox still fills, nobody is told live.
258 let notify = self.env.service("NOTIFY").ok();
Merge branch 'worktree-agent-a5c94df323642630e'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 }
Inbox: the events service tells people what needs them as events arrive263 let sources = inbox::Sources {
264 work: &work,
265 repos: &repos,
266 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)267 notify: notify.as_ref(),
Inbox: the events service tells people what needs them as events arrive268 };
269 inbox::deliver(&self.db, &sources, events).await;
A last-commits walk answers within 2.5 s with what it has, and keeps only finished answers270 Ok(())
271 }
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails272
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 }
A last-commits walk answers within 2.5 s with what it has, and keeps only finished answers302}
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?),
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.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?),
A last-commits walk answers within 2.5 s with what it has, and keeps only finished answers319 "audit_record" => reply(&audit::record(&events.db, args(body)?).await?),
320 "audit_list" => reply(&audit::list(&events.db, args(body)?).await?),
Inbox: the events service tells people what needs them as events arrive321 "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?),
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)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 }
Inbox: threads, reasons, subscriptions and watching338 "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?),
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb97352 "inbox_watchers" => reply(&subscriptions::watchers_count(&events.db, args(body)?).await?),
Inbox: threads, reasons, subscriptions and watching353 "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?),
A last-commits walk answers within 2.5 s with what it has, and keeps only finished answers356 _ => Response::error("Unknown method", 404),
357 }
358}
359
360/// Events from the bus. A batch that fails is retried whole, which is safe
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails361/// 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).
A last-commits walk answers within 2.5 s with what it has, and keeps only finished answers364#[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)]
Merge remote-tracking branch 'origin/claude/cool-newton-7nm8r9' into claude/cool-newton-7nm8r9386async 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 }
A last-commits walk answers within 2.5 s with what it has, and keeps only finished answers398 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();
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails409 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 }
Inbox: the events service tells people what needs them as events arrive417 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 }
A last-commits walk answers within 2.5 s with what it has, and keeps only finished answers422 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}
Merge leftovers: plan activity filtered in the query, pushes counted by account, ghost during the deletion window, profile time zones (identity 0039)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}

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