Skip to content
Sign up

flagon-io/g1t

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

Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays1//! 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).
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays9//!
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//!
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays14//! 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.
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays23
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;
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays31
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};
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays38use 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.
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays57const 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;
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays60
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.
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays62#[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 {
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays64 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?;
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays112 }
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
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays144 })
145 .collect();
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails146 let events: Vec<&Event> = events.iter().collect();
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays147 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();
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays161 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.
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays181 async fn deliver(&self, events: &[Event]) -> Result<()> {
Merge branch 'worktree-agent-ad8a36dfcd4176015' into spend-guardrails182 let mut statements = Vec::with_capacity(events.len() + 1);
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays183 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 }
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays217 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?;
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays221
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;
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays225 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);
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays251 }
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;
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays270 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 }
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays302}
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?),
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays319 "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?),
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays356 _ => 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).
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays364#[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
Audit logs are kept by plan: a week on free, 90 days on the plan, and what staff set for an account in sudo380/// 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.
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays385#[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 }
Audit logs are kept by plan: a week on free, 90 days on the plan, and what staff set for an account in sudo398 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);
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays407 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 sudo408 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 }
Audit logs are kept by plan: a week on free, 90 days on the plan, and what staff set for an account in sudo422 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 }
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays426 Ok(_) => {}
427 Err(error) => worker::console_error!("could not remove old audit entries: {error}"),
428 }
Audit logs are kept by plan: a week on free, 90 days on the plan, and what staff set for an account in sudo429 // 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 }
Spend caps: a monthly budget for comped workspaces and a daily breaker on what g1t pays447}
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.