Skip to content
Sign up

flagon-io/g1t

public
0

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

Mission control counts people's direct pushes to the default branch1//! 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).
Mission control counts people's direct pushes to the default branch9//!
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//!
Mission control counts people's direct pushes to the default branch14//! 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.
Mission control counts people's direct pushes to the default branch23
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;
Mission control counts people's direct pushes to the default branch31
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};
Mission control counts people's direct pushes to the default branch38use 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.
Mission control counts people's direct pushes to the default branch57const 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;
Mission control counts people's direct pushes to the default branch60
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.
Mission control counts people's direct pushes to the default branch62#[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 {
Mission control counts people's direct pushes to the default branch64 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?;
Mission control counts people's direct pushes to the default branch112 }
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
Mission control counts people's direct pushes to the default branch144 })
145 .collect();
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails146 let events: Vec<&Event> = events.iter().collect();
Mission control counts people's direct pushes to the default branch147 send(&js::binding(&self.env, "BUS")?, &events).await
148 }
149
Merge branch 'worktree-agent-ad531323a21f4c975'150 /// Newest first, or with `after` oldest first in the order the log
151 /// wrote them. Callers must have checked that the viewer may see the
Mission control counts people's direct pushes to the default branch152 /// repository asked about.
153 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)154 let (filter, binds) = list_filter(&a);
155 let mut values: Vec<JsValue> = binds
156 .into_iter()
157 .map(|bind| match bind {
158 Bind::Text(text) => JsValue::from(text),
159 Bind::Number(number) => JsValue::from(number),
160 })
161 .collect();
Mission control counts people's direct pushes to the default branch162 values.push(a.limit.unwrap_or(DEFAULT_PAGE).min(MAX_PAGE).into());
Merge branch 'worktree-agent-ad531323a21f4c975'163 let order = if a.after.is_some() { "seq ASC" } else { "id DESC" };
Mission control counts people's direct pushes to the default branch164 let rows = self
165 .db
166 .prepare(format!(
Merge branch 'worktree-agent-ad531323a21f4c975'167 "SELECT * FROM events {filter} ORDER BY {order} LIMIT ?"
Mission control counts people's direct pushes to the default branch168 ))
169 .bind(&values)?
170 .all()
171 .await?
172 .results::<EventRow>()?;
173 Ok(rows.into_iter().map(Event::from).collect())
174 }
175
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails176 /// Writes a batch from the bus to the log, notes who now follows what
177 /// (inbox.rs), hands each subscriber the events it routes, then tells
178 /// the people it concerns.
179 ///
180 /// Everything before the hand-off is safe to repeat. A retried batch
181 /// skips the queues that already have it: when a hand-off fails part
182 /// way, which queues each event reached is written down first.
Mission control counts people's direct pushes to the default branch183 async fn deliver(&self, events: &[Event]) -> Result<()> {
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails184 let mut statements = Vec::with_capacity(events.len() + 1);
Mission control counts people's direct pushes to the default branch185 for event in events {
186 statements.push(
187 self.db
188 .prepare(
189 // Redelivered batches must not duplicate log rows.
Merge branch 'worktree-agent-ad531323a21f4c975'190 // `seq` is the order of writing, which `after` follows
191 // (0010_events_seq.sql): the log has one writer.
192 "INSERT OR IGNORE INTO events (id, type, source, time, repo_id, actor, data, seq)
193 VALUES (?, ?, ?, ?, ?, ?, ?, (SELECT COALESCE(MAX(seq), 0) + 1 FROM events))",
Mission control counts people's direct pushes to the default branch194 )
195 .bind(&[
196 event.id.as_str().into(),
197 event.kind.as_str().into(),
198 event.source.as_str().into(),
199 event.time.as_str().into(),
200 optional(&event.repo_id),
201 optional(&event.actor),
202 serde_json::to_string(&event.data)?.into(),
203 ])?,
204 );
205 }
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails206 let ids: Vec<&str> = events.iter().map(|event| event.id.as_str()).collect();
207 // Read in the same round trip: nearly always nothing.
208 statements.push(
209 self.db
210 .prepare("SELECT event_id, bindings FROM fanout_sent WHERE event_id IN (SELECT value FROM json_each(?))")
211 .bind(&[serde_json::to_string(&ids)?.into()])?,
212 );
213 let results = self.db.batch(statements).await?;
214 let mut sent: HashMap<String, HashSet<String>> = HashMap::new();
215 if let Some(rows) = results.last() {
216 for row in rows.results::<FanoutRow>()? {
217 let bindings: Vec<String> = serde_json::from_str(&row.bindings).unwrap_or_default();
218 sent.insert(row.event_id, bindings.into_iter().collect());
219 }
220 }
Mission control counts people's direct pushes to the default branch221 audit::follow_renames(&self.db, events).await?;
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails222 // Before the hand-off, so its failing does not send the batch to
223 // every queue again.
224 inbox::follow(&self.db, events).await?;
Mission control counts people's direct pushes to the default branch225
226 let bindings: &JsValue = self.env.as_ref();
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails227 let mut reached: Vec<String> = Vec::new();
228 let mut failed = None;
Mission control counts people's direct pushes to the default branch229 for name in Object::keys(bindings.unchecked_ref::<Object>()).iter() {
230 let Some(name) = name
231 .as_string()
232 .filter(|name| name.starts_with(SUBSCRIBER_PREFIX))
233 else {
234 continue;
235 };
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails236 let routed: Vec<&Event> = events
237 .iter()
238 .filter(|event| g1t_contracts::subscribers::routed(&name, &event.kind))
239 .filter(|event| !sent.get(&event.id).is_some_and(|reached| reached.contains(&name)))
240 .collect();
241 if routed.is_empty() {
242 continue;
243 }
244 match send(&js::binding(&self.env, &name)?, &routed).await {
245 Ok(()) => reached.push(name),
246 Err(error) => {
247 worker::console_error!("events: passing {} events to {name} failed: {error}", routed.len());
248 failed = Some(error);
249 }
250 }
251 }
252 if let Some(error) = failed {
253 self.note_reached(events, &sent, &reached).await?;
254 return Err(error);
Mission control counts people's direct pushes to the default branch255 }
Inbox: the events service tells people what needs them as events arrive256 let (work, repos, identity) = (
257 self.env.service("WORK")?,
258 self.env.service("REPOS")?,
259 self.env.service("IDENTITY")?,
260 );
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)261 // Optional: without it the inbox still fills, nobody is told live.
262 let notify = self.env.service("NOTIFY").ok();
Merge branch 'worktree-agent-a5c94df323642630e'263 // Pages open on the repositories: told first, as nobody waits longer.
264 if let Some(notify) = notify.as_ref() {
265 live::tell(notify, events).await;
266 }
Inbox: the events service tells people what needs them as events arrive267 let sources = inbox::Sources {
268 work: &work,
269 repos: &repos,
270 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)271 notify: notify.as_ref(),
Inbox: the events service tells people what needs them as events arrive272 };
273 inbox::deliver(&self.db, &sources, events).await;
Mission control counts people's direct pushes to the default branch274 Ok(())
275 }
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails276
277 /// Writes down which queues each event of a batch has reached, with
278 /// those it had reached before, for the batch's retry.
279 async fn note_reached(
280 &self,
281 events: &[Event],
282 sent: &HashMap<String, HashSet<String>>,
283 reached: &[String],
284 ) -> Result<()> {
285 let now = rfc3339(now_ms());
286 let mut statements = Vec::with_capacity(events.len());
287 for event in events {
288 let mut bindings: Vec<&str> = reached.iter().map(String::as_str).collect();
289 if let Some(before) = sent.get(&event.id) {
290 bindings.extend(before.iter().map(String::as_str));
291 }
292 bindings.sort_unstable();
293 bindings.dedup();
294 statements.push(
295 self.db
296 .prepare(
297 "INSERT INTO fanout_sent (event_id, bindings, created_at) VALUES (?, ?, ?)
298 ON CONFLICT (event_id) DO UPDATE SET bindings = excluded.bindings",
299 )
300 .bind(&[event.id.as_str().into(), serde_json::to_string(&bindings)?.into(), now.as_str().into()])?,
301 );
302 }
303 self.db.batch(statements).await?;
304 Ok(())
305 }
Mission control counts people's direct pushes to the default branch306}
307
308#[event(fetch)]
309async fn fetch(mut request: Request, env: Env, _ctx: Context) -> Result<Response> {
310 let Some(method) = rpc_method(&request) else {
311 return Response::error("Not found", 404);
312 };
313 let body: serde_json::Value = request.json().await?;
314 let events = Events {
315 db: env.d1("DB")?,
316 env,
317 };
318 match method.as_str() {
319 "publish" => reply(&events.publish(args(body)?).await?),
320 "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.321 // Home's digest of a span: counted, by who did it (digest.rs).
322 "activity_digest" => reply(&digest::activity_digest(&events.db, args(body)?).await?),
Mission control counts people's direct pushes to the default branch323 "audit_record" => reply(&audit::record(&events.db, args(body)?).await?),
324 "audit_list" => reply(&audit::list(&events.db, args(body)?).await?),
Inbox: the events service tells people what needs them as events arrive325 "inbox_list" => {
326 let repos = events.env.service("REPOS")?;
327 reply(&inbox::list(&events.db, &repos, args(body)?).await?)
328 }
329 "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)330 "inbox_mark" => {
331 let marked: g1t_contracts::inbox::MarkInboxArgs = args(body)?;
332 let username = marked.username.clone();
333 let changed = inbox::mark(&events.db, marked).await?;
334 // Notify: the new count in every open tab, whoever marked (the site, the API, MCP).
335 if changed > 0
336 && let Ok(notify) = events.env.service("NOTIFY")
337 {
338 inbox::tell_inbox_count(&events.db, &notify, &username).await;
339 }
340 reply(&changed)
341 }
Inbox: threads, reasons, subscriptions and watching342 "inbox_thread" => {
343 let (repos, work) = (events.env.service("REPOS")?, events.env.service("WORK")?);
344 reply(&inbox::thread(&events.db, &repos, &work, args(body)?).await?)
345 }
346 "inbox_subscription" => {
347 let work = events.env.service("WORK")?;
348 reply(&subscriptions::subscription(&events.db, &work, args(body)?).await?)
349 }
350 "inbox_subscribe" => {
351 let work = events.env.service("WORK")?;
352 reply(&subscriptions::subscribe(&events.db, &work, args(body)?).await?)
353 }
354 "inbox_watching" => reply(&subscriptions::watching(&events.db, args(body)?).await?),
355 "inbox_watch" => reply(&subscriptions::watch(&events.db, args(body)?).await?),
Merge branch 'main' into worktree-agent-a69aeabc4b0deeb97356 "inbox_watchers" => reply(&subscriptions::watchers_count(&events.db, args(body)?).await?),
Inbox: threads, reasons, subscriptions and watching357 "inbox_watched" => reply(&subscriptions::watched(&events.db, args(body)?).await?),
358 "inbox_settings" => reply(&subscriptions::settings(&events.db, args(body)?).await?),
359 "inbox_update_settings" => reply(&subscriptions::update_settings(&events.db, args(body)?).await?),
Mission control counts people's direct pushes to the default branch360 _ => Response::error("Unknown method", 404),
361 }
362}
363
364/// Events from the bus. A batch that fails is retried whole, which is safe
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails365/// because the log ignores an event it has seen, and queues that already
366/// have it are skipped (`deliver`). After its retries it goes to the
367/// dead-letter queue (wrangler.jsonc).
Mission control counts people's direct pushes to the default branch368#[event(queue)]
369async fn queue(batch: MessageBatch<Event>, env: Env, _ctx: Context) -> Result<()> {
370 let events = Events {
371 db: env.d1("DB")?,
372 env,
373 };
374 let delivered: Vec<Event> = batch
375 .messages()?
376 .into_iter()
377 .map(|message| message.into_body())
378 .collect();
379 events.deliver(&delivered).await?;
380 batch.ack_all();
381 Ok(())
382}
383
384/// Once a day, old audit entries are removed: first everything older than
385/// any workspace keeps (`AUDIT_MAX_DAYS`, 400 days), then each workspace's
386/// entries older than its own plan keeps, as billing says (7 days free, 90
387/// on the plan, or what staff set). Workspaces with nothing older than the
388/// shortest (`AUDIT_MIN_DAYS`, 7) are left alone.
389#[event(scheduled)]
Merge remote-tracking branch 'origin/claude/cool-newton-7nm8r9' into claude/cool-newton-7nm8r9390async fn scheduled(event: worker::ScheduledEvent, env: Env, _ctx: worker::ScheduleContext) {
391 // The morning cron sends the security digests that are due, and nothing else.
392 if event.cron() == security_digest::DIGEST_CRON {
393 let (Ok(db), Ok(identity)) = (env.d1("DB"), env.service("IDENTITY")) else { return };
394 let notify = env.service("NOTIFY").ok();
395 match security_digest::send_due(&db, &identity, notify.as_ref(), now_ms()).await {
396 Ok(sent) if sent > 0 => worker::console_log!("sent {sent} security digests"),
397 Ok(_) => {}
398 Err(error) => worker::console_error!("security digests not sent: {error}"),
399 }
400 return;
401 }
Mission control counts people's direct pushes to the default branch402 let days = |name: &str, default: u32| {
403 env.var(name)
404 .ok()
405 .and_then(|v| v.to_string().trim().parse::<u32>().ok())
406 .filter(|days| *days > 0)
407 .unwrap_or(default)
408 };
409 let max_days = days("AUDIT_MAX_DAYS", audit::DEFAULT_MAX_DAYS);
410 let min_days = days("AUDIT_MIN_DAYS", audit::DEFAULT_MIN_DAYS).min(max_days);
411 let Ok(db) = env.d1("DB") else { return };
412 let now = now_ms();
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails413 let cutoff = rfc3339(now.saturating_sub(FANOUT_KEEP_MS));
414 let purged = match db.prepare("DELETE FROM fanout_sent WHERE created_at < ?").bind(&[cutoff.into()]) {
415 Ok(statement) => statement.run().await.map(|_| ()),
416 Err(error) => Err(error),
417 };
418 if let Err(error) = purged {
419 worker::console_error!("could not remove old fan-out records: {error}");
420 }
Inbox: the events service tells people what needs them as events arrive421 match inbox::purge(&db, now).await {
422 Ok(removed) if removed > 0 => worker::console_log!("removed {removed} old inbox items"),
423 Ok(_) => {}
424 Err(error) => worker::console_error!("could not remove old inbox items: {error}"),
425 }
Mission control counts people's direct pushes to the default branch426 match audit::purge(&db, &audit::keep_from(now, max_days), 20).await {
427 Ok(removed) if removed > 0 => {
428 worker::console_log!("removed {removed} audit entries older than {max_days} days")
429 }
430 Ok(_) => {}
431 Err(error) => worker::console_error!("could not remove old audit entries: {error}"),
432 }
433 // Without billing nobody's plan is known, so nothing younger than the
434 // ceiling is removed.
435 let billing = match env.service("BILLING") {
436 Ok(billing) => billing,
437 Err(error) => {
438 worker::console_error!(
439 "audit entries kept past their plan's days: no billing: {error}"
440 );
441 return;
442 }
443 };
444 match audit::purge_by_plan(&db, &billing, now, min_days, max_days).await {
445 Ok(removed) if removed > 0 => {
446 worker::console_log!("removed {removed} audit entries older than their plan keeps")
447 }
448 Ok(_) => {}
449 Err(error) => worker::console_error!("audit entries kept past their plan's days: {error}"),
450 }
451}
Merge leftovers: plan activity filtered in the query, pushes counted by account, ghost during the deletion window, profile time zones (identity 0039)452
Merge branch 'worktree-agent-ad531323a21f4c975'453/// The condition of an `after` cursor, binding its id twice.
454const AFTER: &str = concat!(
455 "seq > COALESCE(",
456 "(SELECT seq FROM events WHERE id = ?), ",
457 "(SELECT seq FROM events WHERE id > ? AND seq IS NOT NULL ORDER BY id LIMIT 1) - 1, ",
458 "(SELECT MAX(seq) FROM events), 0)"
459);
460
Merge leftovers: plan activity filtered in the query, pushes counted by account, ghost during the deletion window, profile time zones (identity 0039)461/// A value bound to a `?` in [`list_filter`]'s clause.
462#[derive(Debug, PartialEq)]
463enum Bind {
464 Text(String),
465 Number(f64),
466}
467
468/// The `WHERE` clause `list` reads with, and what it binds, in order.
469fn list_filter(a: &ListArgs) -> (String, Vec<Bind>) {
470 let mut conditions = Vec::new();
471 let mut values = Vec::new();
472 if let Some(repo_id) = &a.repo_id {
473 conditions.push("repo_id = ?".to_owned());
474 values.push(Bind::Text(repo_id.clone()));
475 }
476 if !a.types.is_empty() {
477 let marks = vec!["?"; a.types.len()].join(", ");
478 conditions.push(format!("type IN ({marks})"));
479 values.extend(a.types.iter().map(|kind| Bind::Text(kind.clone())));
480 } else {
481 // What CI and integrations report on commits goes to webhooks,
482 // and is read from each commit's checks; a timeline asked for
483 // everything would be little else on a busy repository.
484 let marks = vec!["?"; REPORTING.len()].join(", ");
485 conditions.push(format!("type NOT IN ({marks})"));
486 values.extend(REPORTING.iter().map(|kind| Bind::Text((*kind).to_owned())));
487 }
488 if let Some(actor) = &a.actor {
489 conditions.push("actor = ?".to_owned());
490 values.push(Bind::Text(actor.clone()));
491 }
492 if !a.numbers.is_empty() {
493 // An issue or pull request's own events name it as `number`; a
494 // comment, review or link on it names it as `issue`.
495 let marks = vec!["?"; a.numbers.len()].join(", ");
496 conditions.push(format!(
497 "(json_extract(data, '$.number') IN ({marks}) OR json_extract(data, '$.issue') IN ({marks}))"
498 ));
499 for _ in 0..2 {
500 values.extend(a.numbers.iter().map(|number| Bind::Number(f64::from(*number))));
501 }
502 }
503 if let Some(since) = &a.since {
504 conditions.push("time >= ?".to_owned());
505 values.push(Bind::Text(since.clone()));
506 }
507 if let Some(before) = &a.before {
508 conditions.push("id < ?".to_owned());
509 values.push(Bind::Text(before.clone()));
510 }
Merge branch 'worktree-agent-ad531323a21f4c975'511 if let Some(after) = &a.after {
512 // What the log wrote after the cursor's event. A cursor the log
513 // does not have (yet: an id from a webhook can arrive before its
514 // row is written) starts at the first event with a later id, and
515 // one past them all at the end of the log.
516 conditions.push(AFTER.to_owned());
517 values.push(Bind::Text(after.clone()));
518 values.push(Bind::Text(after.clone()));
519 }
Merge leftovers: plan activity filtered in the query, pushes counted by account, ghost during the deletion window, profile time zones (identity 0039)520 let filter = if conditions.is_empty() {
521 String::new()
522 } else {
523 format!("WHERE {}", conditions.join(" AND "))
524 };
525 (filter, values)
526}
527
528#[cfg(test)]
529mod tests {
530 use super::*;
531
532 #[test]
533 fn a_plain_list_leaves_out_what_is_reported_on_commits() {
534 let (filter, binds) = list_filter(&ListArgs { repo_id: Some("rep_1".into()), ..ListArgs::default() });
535 assert!(filter.starts_with("WHERE repo_id = ? AND type NOT IN ("));
536 assert_eq!(binds[0], Bind::Text("rep_1".into()));
537 assert_eq!(binds.len(), 1 + REPORTING.len());
538 }
539
540 #[test]
541 fn numbers_match_an_item_or_what_is_said_on_it_since_a_time() {
542 let (filter, binds) = list_filter(&ListArgs {
543 repo_id: Some("rep_1".into()),
544 types: vec!["issue.opened".into()],
545 numbers: vec![4, 9],
546 since: Some("2026-10-01T00:00:00Z".into()),
547 actor: Some("usr_g1t_agent".into()),
548 ..ListArgs::default()
549 });
550 assert_eq!(
551 filter,
552 "WHERE repo_id = ? AND type IN (?) AND actor = ? AND \
553 (json_extract(data, '$.number') IN (?, ?) OR json_extract(data, '$.issue') IN (?, ?)) AND time >= ?"
554 );
555 assert_eq!(
556 binds,
557 vec![
558 Bind::Text("rep_1".into()),
559 Bind::Text("issue.opened".into()),
560 Bind::Text("usr_g1t_agent".into()),
561 Bind::Number(4.0),
562 Bind::Number(9.0),
563 Bind::Number(4.0),
564 Bind::Number(9.0),
565 Bind::Text("2026-10-01T00:00:00Z".into()),
566 ]
567 );
568 }
Merge branch 'worktree-agent-ad531323a21f4c975'569
570 #[test]
571 fn an_after_cursor_reads_what_the_log_wrote_later_with_every_other_filter() {
572 let (filter, binds) = list_filter(&ListArgs {
573 repo_id: Some("rep_1".into()),
574 types: vec!["work.claimed".into(), "integration.failed".into()],
575 numbers: vec![7],
576 since: Some("2026-10-12T00:00:00Z".into()),
577 after: Some("evt_9".into()),
578 ..ListArgs::default()
579 });
580 assert_eq!(
581 filter,
582 format!(
583 "WHERE repo_id = ? AND type IN (?, ?) AND \
584 (json_extract(data, '$.number') IN (?) OR json_extract(data, '$.issue') IN (?)) AND time >= ? AND {AFTER}"
585 )
586 );
587 // The cursor's id is bound once to find its place, and once for a
588 // cursor the log does not have.
589 assert_eq!(&binds[binds.len() - 2..], &[Bind::Text("evt_9".into()), Bind::Text("evt_9".into())]);
590 assert_eq!(binds.len(), 8);
591 assert_eq!(AFTER.matches('?').count(), 2);
592 }
Merge leftovers: plan activity filtered in the query, pushes counted by account, ghost during the deletion window, profile time zones (identity 0039)593}

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