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

Commit

A repository's events can be followed: list_events takes after, an event id, and returns the events the log wrote after it oldest first, with types and numbers (comma lists in a query, arrays in a body or MCP call), since and limit (at most 200) alongside, and with wait (seconds, at most 25) a call that finds nothing holds on, asking the events service about once a second, until a matching event is written or the wait ends. The log numbers events in the order it writes them (seq, migration 0010), and after follows that order rather than ids, which are made where events are published and can reach the log out of order, so a follower never skips one; a cursor the log does not have starts at the next id it does. Holding a request costs the Worker wall time, not CPU time, and at most 26 subrequests. The docs, the OpenAPI document and the CLI's operation table describe the new parameters.

syntaqxcommitted Parent85c083aBrowse files
12 files+373−210/12 viewed
+231−0
1+//! `list_events`: a repository's timeline, and following it.
2+//!
3+//! Without `after` it is the timeline, newest first, paged back with
4+//! `before`. With `after` (an event id) it is what the log wrote after that
5+//! event, oldest first, and the last event's id is the cursor for the next
6+//! call. With `after` and `wait`, a call that finds nothing holds on until
7+//! something matching is written or `wait` seconds pass, asking the events
8+//! service about once a second, so an agent with no public address can
9+//! follow a repository with one request at a time instead of a webhook.
10+//!
11+//! Holding a request costs a Worker wall time, not CPU time: between polls
12+//! it awaits a timer, which is not billed or limited as CPU. Each poll is
13+//! one subrequest, so a call makes at most `MAX_WAIT_SECONDS` + 1 of them,
14+//! well under the per-request limit, and the hold stays far below the time
15+//! clients and proxies give an idle response.
16+
17+use g1t_contracts::events::{Event, ListArgs};
18+use serde_json::Value;
19+use worker::{Fetcher, Result};
20+
21+/// The longest a call holds on, in seconds.
22+pub const MAX_WAIT_SECONDS: u64 = 25;
23+/// How often a held call asks again.
24+const POLL_MS: u64 = 1_000;
25+/// The most events one call returns; the events service's own page.
26+const MAX_LIMIT: u32 = 200;
27+
28+/// What a `list_events` call asks for, read from its input.
29+#[derive(Debug, Default, PartialEq)]
30+pub struct Asked {
31+ pub after: Option<String>,
32+ pub before: Option<String>,
33+ pub types: Vec<String>,
34+ pub numbers: Vec<u32>,
35+ pub since: Option<String>,
36+ pub limit: Option<u32>,
37+ /// Milliseconds to hold on; zero without `after`.
38+ pub wait_ms: u64,
39+}
40+
41+/// A list given as an array or as one comma-separated string.
42+fn list(value: &Value) -> Vec<String> {
43+ let items: Vec<String> = match value {
44+ Value::String(text) => text.split(',').map(str::to_owned).collect(),
45+ Value::Array(items) => items
46+ .iter()
47+ .map(|item| match item {
48+ Value::String(text) => text.clone(),
49+ other => other.to_string(),
50+ })
51+ .collect(),
52+ Value::Number(number) => vec![number.to_string()],
53+ _ => Vec::new(),
54+ };
55+ items
56+ .into_iter()
57+ .map(|item| item.trim().trim_start_matches('#').to_owned())
58+ .filter(|item| !item.is_empty())
59+ .collect()
60+}
61+
62+fn text(input: &Value, key: &str) -> Option<String> {
63+ input[key]
64+ .as_str()
65+ .map(str::trim)
66+ .filter(|value| !value.is_empty())
67+ .map(str::to_owned)
68+}
69+
70+fn whole(value: &Value) -> Option<u64> {
71+ match value {
72+ Value::Number(number) => number.as_u64().or_else(|| number.as_f64().filter(|n| *n >= 0.0).map(|n| n as u64)),
73+ Value::String(digits) => digits.trim().parse().ok(),
74+ _ => None,
75+ }
76+}
77+
78+/// Reads a call's input, or says what is wrong with it.
79+pub fn asked(input: &Value) -> std::result::Result<Asked, String> {
80+ let mut numbers = Vec::new();
81+ for item in list(&input["numbers"]) {
82+ match item.parse::<u32>() {
83+ Ok(number) if number > 0 => numbers.push(number),
84+ _ => return Err(format!("numbers are issue and pull request numbers, and {item} is not one.")),
85+ }
86+ }
87+ numbers.sort_unstable();
88+ numbers.dedup();
89+ let mut types = list(&input["types"]);
90+ types.sort();
91+ types.dedup();
92+ let limit = match &input["limit"] {
93+ Value::Null => None,
94+ value => match whole(value) {
95+ Some(limit) if limit > 0 => Some(u32::try_from(limit).unwrap_or(MAX_LIMIT).min(MAX_LIMIT)),
96+ _ => return Err("limit is a whole number from 1 to 200.".to_owned()),
97+ },
98+ };
99+ let after = text(input, "after");
100+ let wait = match &input["wait"] {
101+ Value::Null => 0,
102+ value => match whole(value) {
103+ Some(seconds) => seconds.min(MAX_WAIT_SECONDS),
104+ None => return Err(format!("wait is a number of seconds, at most {MAX_WAIT_SECONDS}.")),
105+ },
106+ };
107+ let before = text(input, "before");
108+ if after.is_some() && before.is_some() {
109+ return Err("Give after to follow the timeline or before to page back through it, not both.".to_owned());
110+ }
111+ Ok(Asked {
112+ wait_ms: if after.is_some() { wait * 1_000 } else { 0 },
113+ after,
114+ before,
115+ types,
116+ numbers,
117+ since: text(input, "since"),
118+ limit,
119+ })
120+}
121+
122+impl Asked {
123+ fn args(&self, repo_id: &str) -> ListArgs {
124+ ListArgs {
125+ repo_id: Some(repo_id.to_owned()),
126+ types: self.types.clone(),
127+ before: self.before.clone(),
128+ after: self.after.clone(),
129+ numbers: self.numbers.clone(),
130+ since: self.since.clone(),
131+ limit: self.limit,
132+ ..ListArgs::default()
133+ }
134+ }
135+}
136+
137+/// How long to sleep before asking again, at `now`, for a call that holds
138+/// on until `deadline`: `None` when it is time to answer with nothing.
139+/// The last poll falls on the deadline itself.
140+pub fn next_poll(now: u64, deadline: u64) -> Option<u64> {
141+ if now >= deadline {
142+ None
143+ } else {
144+ Some(POLL_MS.min(deadline - now))
145+ }
146+}
147+
148+/// The events of the repository `repo_id` as asked, holding on for them
149+/// when asked to.
150+pub async fn list_events(events: &Fetcher, repo_id: &str, asked: &Asked) -> Result<Vec<Event>> {
151+ let args = asked.args(repo_id);
152+ let deadline = g1t_kit::now_ms() + asked.wait_ms;
153+ loop {
154+ let found: Vec<Event> = g1t_kit::call(events, "list", &args).await?;
155+ if !found.is_empty() {
156+ return Ok(found);
157+ }
158+ match next_poll(g1t_kit::now_ms(), deadline) {
159+ Some(ms) => worker::Delay::from(std::time::Duration::from_millis(ms)).await,
160+ None => return Ok(found),
161+ }
162+ }
163+}
164+
165+#[cfg(test)]
166+mod tests {
167+ use super::*;
168+ use serde_json::json;
169+
170+ #[test]
171+ fn lists_come_as_commas_or_arrays() {
172+ let a = asked(&json!({
173+ "types": "work.claimed, integration.failed,,work.claimed",
174+ "numbers": "12,#7, 12",
175+ }))
176+ .unwrap();
177+ assert_eq!(a.types, vec!["integration.failed", "work.claimed"]);
178+ assert_eq!(a.numbers, vec![7, 12]);
179+ let b = asked(&json!({ "types": ["push"], "numbers": [3, "4"] })).unwrap();
180+ assert_eq!(b.types, vec!["push"]);
181+ assert_eq!(b.numbers, vec![3, 4]);
182+ assert!(asked(&json!({ "numbers": "seven" })).is_err());
183+ assert!(asked(&json!({ "numbers": [0] })).is_err());
184+ }
185+
186+ #[test]
187+ fn wait_holds_only_a_follow_and_never_past_25_seconds() {
188+ assert_eq!(asked(&json!({ "wait": 10 })).unwrap().wait_ms, 0, "nothing to wait for without after");
189+ assert_eq!(asked(&json!({ "after": "evt_1", "wait": "10" })).unwrap().wait_ms, 10_000);
190+ assert_eq!(asked(&json!({ "after": "evt_1", "wait": 600 })).unwrap().wait_ms, 25_000);
191+ assert_eq!(asked(&json!({ "after": "evt_1" })).unwrap().wait_ms, 0);
192+ assert!(asked(&json!({ "after": "evt_1", "wait": "soon" })).is_err());
193+ assert!(asked(&json!({ "after": "evt_1", "wait": -1 })).is_err());
194+ }
195+
196+ #[test]
197+ fn limits_are_kept_to_a_page() {
198+ assert_eq!(asked(&json!({ "limit": "500" })).unwrap().limit, Some(200));
199+ assert_eq!(asked(&json!({ "limit": 20 })).unwrap().limit, Some(20));
200+ assert_eq!(asked(&json!({})).unwrap().limit, None);
201+ assert!(asked(&json!({ "limit": 0 })).is_err());
202+ }
203+
204+ #[test]
205+ fn after_and_before_do_not_mix() {
206+ assert!(asked(&json!({ "after": "evt_2", "before": "evt_1" })).is_err());
207+ let a = asked(&json!({ "after": "evt_2", "since": "2026-10-12T00:00:00Z" })).unwrap();
208+ let args = a.args("rep_1");
209+ assert_eq!(args.after.as_deref(), Some("evt_2"));
210+ assert_eq!(args.since.as_deref(), Some("2026-10-12T00:00:00Z"));
211+ assert_eq!(args.repo_id.as_deref(), Some("rep_1"));
212+ }
213+
214+ #[test]
215+ fn polls_come_about_once_a_second_and_the_last_on_the_deadline() {
216+ let start = 1_000_000;
217+ let deadline = start + 25_000;
218+ let mut now = start;
219+ let mut polls = 1; // the first, before any sleep
220+ while let Some(ms) = next_poll(now, deadline) {
221+ assert!(ms > 0 && ms <= 1_000);
222+ now += ms;
223+ polls += 1;
224+ }
225+ assert_eq!(now, deadline);
226+ assert_eq!(polls, 26, "one call makes at most 26 polls");
227+ assert_eq!(next_poll(5, 5), None);
228+ assert_eq!(next_poll(5, 4), None, "a slow poll past the deadline answers");
229+ assert_eq!(next_poll(0, 300), Some(300));
230+ }
231+}
+1−0
2525 mod oidc;
2626 mod packages;
2727 mod folios;
28+mod follow;
2829 mod people;
2930 mod openapi;
3031 mod pins;
+27−13
1111 };
1212 use g1t_contracts::codeowners::CodeOwnersErrorsArgs;
1313 use g1t_contracts::identity::AgentScope;
14−use g1t_contracts::events::{Event, ListArgs as ListEventsArgs};
14+use g1t_contracts::events::Event;
1515 use g1t_contracts::identity::{CreateWorkspaceArgs, UpdateWorkspaceArgs, Workspace};
1616 use g1t_contracts::repos::{CreateArgs, GetArgs, ListArgs as ListReposArgs, Repo, RepoPath};
1717 use g1t_contracts::teams::{
14661466 "Land a pull request on its base, the branch it merges into (the default branch unless it names another). Merging needs the Write role or higher, and only once it is marked ready and it meets every rule that holds for its base (see rules and required_checks on get_pull_request: approvals, checks, deployments, merge windows and the rest, from the repository's and its workspace's rulesets); the refusal names the first rule not met. With ignore_checks, someone who may merge can bypass required checks where the rule allows it; with bypass_rules, someone a ruleset lists as a bypass actor merges past its rules, and it is recorded. Merging into the default branch resolves the issue it was made for: the issue closes recording this pull request, and the other pull requests still in progress for that issue close as superseded; merging into another branch leaves the issue open. Where the repository has a merge queue, a pull request into the default branch joins the queue instead of landing at once. If its base has moved since the pull request was opened, it is brought up to date first and lands when that is done; a repository that requires pull requests into its default branch to be up to date refuses instead, so pull the base into its fork or branch, push, and merge again. Check status in the result to see whether it has landed."
14671467 }
14681468 Op::ListEvents => {
1469− "The timeline of a repository: pushes, issues, pull requests, comments and session activity, newest first."
1469+ "The timeline of a repository: pushes, issues, pull requests, comments, sessions, claims and integration results, newest first; page back with before. To follow it, give after (an event id): the events written after that one come oldest first, and the last one's id is the next call's after. With after and wait (seconds, at most 25), a call that finds nothing holds on until a matching event is written or the wait ends, then returns what it has, perhaps nothing. Narrow it with types, numbers (issues and pull requests, matching events about them and comments on them), since and limit (50 by default, at most 200)."
14701470 }
14711471 Op::ListIntegrations => {
14721472 "A workspace's integrations: its AI providers, the alert sources that open issues (Sentry, Datadog, webhooks), and the trackers whose tickets agents can read (Jira, Linear). Secrets are never returned. Members only."
24322432 Op::ListEvents => object(
24332433 json!({
24342434 "repo": repo_schema(),
2435− "before": { "type": "string", "description": "Event id to page back from." },
2435+ "before": { "type": "string", "description": "Event id to page back from: older events, newest first." },
2436+ "after": { "type": "string", "description": "Event id to follow from: the events written after it, oldest first. Not with before." },
2437+ "types": {
2438+ "type": "array",
2439+ "items": { "type": "string" },
2440+ "description": "Only these event types (work.claimed, integration.failed, ...). In a query string, a comma-separated list.",
2441+ },
2442+ "numbers": {
2443+ "type": "array",
2444+ "items": { "type": "integer" },
2445+ "description": "Only events about these issues or pull requests, and comments, reviews and links on them. In a query string, a comma-separated list.",
2446+ },
2447+ "since": { "type": "string", "description": "Only events at or after this time (RFC 3339)." },
2448+ "limit": { "type": "integer", "minimum": 1, "maximum": 200, "description": "How many events at most; 50 by default." },
2449+ "wait": {
2450+ "type": "integer",
2451+ "minimum": 0,
2452+ "maximum": 25,
2453+ "description": "With after: seconds to hold on for a matching event when there is none yet. At most 25.",
2454+ },
24362455 }),
24372456 &["repo"],
24382457 ),
47504769 .await
47514770 }
47524771 Op::ListEvents => {
4772+ let asked = match crate::follow::asked(input) {
4773+ Ok(asked) => asked,
4774+ Err(message) => return failed(FailureCode::Invalid, &message),
4775+ };
47534776 let found: Outcome<Repo> = call(
47544777 repos,
47554778 "get",
47634786 Outcome::Ok(repo) => repo,
47644787 Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
47654788 };
4766− let timeline: Vec<Event> = g1t_kit::call(
4767− events,
4768− "list",
4769− &ListEventsArgs {
4770− repo_id: Some(repo.id),
4771− before: optional_text(input, "before"),
4772− ..ListEventsArgs::default()
4773− },
4774− )
4775− .await?;
4789+ let timeline: Vec<Event> = crate::follow::list_events(events, &repo.id, &asked).await?;
47764790 ok(&timeline)
47774791 }
47784792 // Who has access: identity decides, from the repository as the
+1−1
946946 }
947947 }
948948 ],
949− "notes": "Returns 50 events a page. To page back, send the `id` of the last event you have as `before`. `actor` is a user id, and `data` depends on `type`."
949+ "notes": "Returns 50 events a page, newest first. To page back, send the `id` of the last event you have as `before`. `actor` is a user id, and `data` depends on `type`.\n\nTo follow a repository without a webhook, send `after` instead: the events written after that event come oldest first, and the last one's `id` is your next `after`. Add `wait` (seconds, at most 25) and a call that finds nothing holds on until a matching event is written or the time is up, then returns what it has, which may be an empty list; call again with the same `after`. Events come in the order g1t wrote them, so none is skipped when two arrive close together.\n\n```sh\ncurl -H \"Authorization: Bearer $G1T_TOKEN\" \\\n \"https://api.g1t.sh/repos/acme/rocket/events?after=evt_01m43smh3ve52vkcna207vqctf&types=work.claimed,integration.failed&numbers=12&wait=25\"\n```\n\n`types` and `numbers` are comma-separated lists in a query string, arrays in a body or an MCP call. `numbers` matches events about those issues and pull requests (`data.number`) and comments, reviews and links on them (`data.issue`). `since` is an RFC 3339 time. `after` and `before` cannot be given together."
950950 },
951951 "list_issues": {
952952 "query": {
+30−1
530530 "GET",
531531 "/repos/:owner/:name/events",
532532 Op::ListEvents,
533− &[("before", "before")],
533+ &[
534+ ("before", "before"),
535+ ("after", "after"),
536+ ("types", "types"),
537+ ("numbers", "numbers"),
538+ ("since", "since"),
539+ ("limit", "limit"),
540+ ("wait", "wait"),
541+ ],
534542 ),
535543 route("GET", "/repos/:owner/:name/labels", Op::ListLabels, &[]),
536544 route("POST", "/repos/:owner/:name/labels", Op::CreateLabel, &[]),
13351343 }
13361344
13371345 #[test]
1346+ fn events_are_followed_with_a_cursor_filters_and_a_wait() {
1347+ let query = [
1348+ ("after".to_owned(), "evt_1".to_owned()),
1349+ ("types".to_owned(), "work.claimed,integration.failed".to_owned()),
1350+ ("numbers".to_owned(), "12,14".to_owned()),
1351+ ("wait".to_owned(), "25".to_owned()),
1352+ ("limit".to_owned(), "20".to_owned()),
1353+ ("since".to_owned(), "2026-10-12T00:00:00Z".to_owned()),
1354+ ];
1355+ let (route, input) = resolve("GET", "/repos/acme/rocket/events", &query, Value::Null).unwrap();
1356+ assert_eq!(route.op, Op::ListEvents);
1357+ let asked = crate::follow::asked(&input).unwrap();
1358+ assert_eq!(asked.after.as_deref(), Some("evt_1"));
1359+ assert_eq!(asked.types, vec!["integration.failed", "work.claimed"]);
1360+ assert_eq!(asked.numbers, vec![12, 14]);
1361+ assert_eq!(asked.wait_ms, 25_000);
1362+ assert_eq!(asked.limit, Some(20));
1363+ assert_eq!(asked.since.as_deref(), Some("2026-10-12T00:00:00Z"));
1364+ }
1365+
1366+ #[test]
13381367 fn notifications_are_addressed_as_threads_and_by_issue() {
13391368 let (route, input) = resolve("DELETE", "/notifications/threads/ntf_1", &[], Value::Null).unwrap();
13401369 assert_eq!(route.op, Op::MarkThreadDone);
+9−0
169169 refused. To receive webhooks on your own machine while you build, use a
170170 tunnel such as Cloudflare Tunnel.
171171
172+## Without a public address
173+
174+An agent on a laptop, or anything else that cannot take requests from the
175+internet, can follow a repository's events instead:
176+[`list_events`](/reference/api/repositories/list-events/) with `after` and
177+`wait` holds each request until an event is written, so one request at a
178+time hears about events within a second or so. `g1t work events --follow`
179+does this for you ([The g1t CLI](/guides/cli/#following-events)).
180+
172181 ## From the API
173182
174183 The same routes, and the MCP server's `webhook` tool, work for a
+1−1
269269 | [`create_milestone`](/reference/api/labels-and-milestones/create-milestone/) | Create a milestone with `title`, `description` and `due_on` (`YYYY-MM-DD`). Write role. | `repo`, `title` | `issues:write` |
270270 | [`update_milestone`](/reference/api/labels-and-milestones/update-milestone/) | Change its `title`, `description`, `due_on` (`""` clears it) or `state` (`open` or `closed`). Write role. | `repo`, `milestone` | `issues:write` |
271271 | [`delete_milestone`](/reference/api/labels-and-milestones/delete-milestone/) | Delete a milestone; what was in it is in none. Write role. | `repo`, `milestone` | `issues:write` |
272−| [`list_events`](/reference/api/repositories/list-events/) | Its timeline, newest first. `before` pages back. | `repo` | `repo:read` |
272+| [`list_events`](/reference/api/repositories/list-events/) | Its timeline, newest first; `before` pages back. To follow it, give `after` (an event id): the events written after it come oldest first, and the last one's `id` is the next `after`. With `wait` (seconds, at most 25) a call that finds nothing holds on until a matching event is written. `types` and `numbers` (arrays) narrow it, with `since` and `limit` (at most 200). | `repo` | `repo:read` |
273273 | [`languages`](/reference/api/repository-insights/get-languages/) | Its [languages](/guides/managing-repositories/#languages) by bytes, each with `color` and `percent`, for the default branch's `commit`; `pending` while it is first read. | `repo` | `repo:read` |
274274 | [`contributors`](/reference/api/repository-insights/list-contributors/) | Its [contributors](/guides/managing-repositories/#contributors): `kind` (`user`, `g1t` or `author`), `commits` and `weeks`, and the repository's commits by week. | `repo` | `repo:read` |
275275 | [`license`](/reference/api/repository-insights/get-license/) | The license its `LICENSE` file holds: `spdx_id`, `name`, `path`. | `repo` | `repo:read` |
+0−0

Binary or large file; its contents are not shown.

+8−1
564564 pub data: serde_json::Value,
565565 }
566566
567−/// `list`: events from the log, newest first. Returns `Vec<Event>`.
567+/// `list`: events from the log, newest first, or with `after` oldest first.
568+/// Returns `Vec<Event>`.
568569 #[derive(Debug, Default, Serialize, serde::Deserialize)]
569570 #[serde(rename_all = "camelCase")]
570571 pub struct ListArgs {
576577 /// Only events older than this event id.
577578 #[serde(default)]
578579 pub before: Option<String>,
580+ /// Only events the log recorded after this event id, and then oldest
581+ /// first: a cursor to follow the log with. Events are ordered by when
582+ /// the log wrote them, not by their ids, so one published a moment
583+ /// before another but written after it is not skipped.
584+ #[serde(default)]
585+ pub after: Option<String>,
579586 /// Only events by this account id.
580587 #[serde(default)]
581588 pub actor: Option<String>,
+6−0
37213721 Param { name: "owner", location: In::Path, kind: "string", required: true, values: &[] },
37223722 Param { name: "name", location: In::Path, kind: "string", required: true, values: &[] },
37233723 Param { name: "before", location: In::Query, kind: "string", required: false, values: &[] },
3724+ Param { name: "after", location: In::Query, kind: "string", required: false, values: &[] },
3725+ Param { name: "types", location: In::Query, kind: "array", required: false, values: &[] },
3726+ Param { name: "numbers", location: In::Query, kind: "array", required: false, values: &[] },
3727+ Param { name: "since", location: In::Query, kind: "string", required: false, values: &[] },
3728+ Param { name: "limit", location: In::Query, kind: "integer", required: false, values: &[] },
3729+ Param { name: "wait", location: In::Query, kind: "integer", required: false, values: &[] },
37243730 ],
37253731 body: &[],
37263732 },
+10−0
1+-- The order the log wrote events in. Ids are made when an event is
2+-- published, on whichever Worker published it, and events reach the log
3+-- through a queue, so an event can be written after one with a later id.
4+-- Following the log with an `after` cursor reads by this instead: what is
5+-- written later always has a higher number, so a follower never skips one.
6+-- Each insert takes the highest so far plus one; the log has one writer
7+-- at a time. Events written before this have none and are not followed.
8+ALTER TABLE events ADD COLUMN seq INTEGER;
9+CREATE INDEX IF NOT EXISTS events_seq ON events (seq);
10+CREATE INDEX IF NOT EXISTS events_repo_seq ON events (repo_id, seq);
+49−4
147147 send(&js::binding(&self.env, "BUS")?, &events).await
148148 }
149149
150− /// Newest first. Callers must have checked that the viewer may see the
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
151152 /// repository asked about.
152153 async fn list(&self, a: ListArgs) -> Result<Vec<Event>> {
153154 let (filter, binds) = list_filter(&a);
159160 })
160161 .collect();
161162 values.push(a.limit.unwrap_or(DEFAULT_PAGE).min(MAX_PAGE).into());
163+ let order = if a.after.is_some() { "seq ASC" } else { "id DESC" };
162164 let rows = self
163165 .db
164166 .prepare(format!(
165− "SELECT * FROM events {filter} ORDER BY id DESC LIMIT ?"
167+ "SELECT * FROM events {filter} ORDER BY {order} LIMIT ?"
166168 ))
167169 .bind(&values)?
168170 .all()
185187 self.db
186188 .prepare(
187189 // Redelivered batches must not duplicate log rows.
188− "INSERT OR IGNORE INTO events (id, type, source, time, repo_id, actor, data)
189− VALUES (?, ?, ?, ?, ?, ?, ?)",
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))",
190194 )
191195 .bind(&[
192196 event.id.as_str().into(),
446450 }
447451 }
448452
453+/// The condition of an `after` cursor, binding its id twice.
454+const 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+
449461 /// A value bound to a `?` in [`list_filter`]'s clause.
450462 #[derive(Debug, PartialEq)]
451463 enum Bind {
496508 conditions.push("id < ?".to_owned());
497509 values.push(Bind::Text(before.clone()));
498510 }
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+ }
499520 let filter = if conditions.is_empty() {
500521 String::new()
501522 } else {
545566 ]
546567 );
547568 }
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+ }
548593 }

Comments

Sign in to comment.