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.

gitagentscloudflarerustforgemcp
362 linesCodeBlameRaw
1//! Issues became work items (migrations/0013_work_items.sql): what this
2//! service kept by a repository's issue number is rewritten to the work
3//! item it became when the work service announces where each went
4//! (`work.migrated`, a page of `MigratedItem`s at a time).
5//!
6//! Each page is recorded in `work_migrated`, then:
7//!
8//! - inbox threads `<repo_id>#<number>` become `item:<item_id>`, with the
9//! work item's workspace and number; a person who already has the work
10//! item's thread keeps that one, with the old one's history;
11//! - subscriptions follow their threads the same way;
12//! - the log's `issue.*` events become `work_item.*`, and its events about
13//! an issue (comments on it, and `issue` / `otherIssue` in coordination
14//! and pull request events) name the work item and its number.
15//!
16//! Every statement leaves what it already rewrote alone, so a page
17//! delivered again, or out of order, changes nothing more. The cron
18//! catches up on pages that never arrived ([`catch_up`]).
19
20use g1t_contracts::events::{Event, WorkMigrated};
21use worker::{D1Database, D1PreparedStatement, Fetcher, Result};
22
23/// The issue events of the log, by their old names.
24pub const ISSUE_EVENTS: [&str; 9] = [
25 "issue.opened",
26 "issue.updated",
27 "issue.assigned",
28 "issue.labeled",
29 "issue.unlabeled",
30 "issue.milestoned",
31 "issue.demilestoned",
32 "issue.closed",
33 "issue.reopened",
34];
35
36/// The page's (`?1`) issues' events and comments, by repository and old
37/// number: the `type` list is `events_issue_number`'s (migration 0015),
38/// word for word, so SQLite reads that index.
39macro_rules! page_events_by_number {
40 () => {
41 "SELECT e.id FROM work_migrated m JOIN events e INDEXED BY events_issue_number
42 ON e.repo_id = m.repo_id AND json_extract(e.data, '$.number') = m.legacy_number
43 AND e.type IN ('issue.opened', 'issue.updated', 'issue.assigned', 'issue.labeled', 'issue.unlabeled', 'issue.milestoned',
44 'issue.demilestoned', 'issue.closed', 'issue.reopened', 'comment.created', 'comment.edited', 'comment.deleted')
45 WHERE m.event_id = ?1"
46 };
47}
48
49/// What one page does, in order: `?1` the event's id, `?2` its items as
50/// JSON, `?3` [`ISSUE_EVENTS`] as JSON.
51pub const MOVED: [&str; 12] = [
52 // The page, kept.
53 "INSERT OR REPLACE INTO work_migrated (repo_id, legacy_number, workspace, number, item_id, event_id)
54 SELECT json_extract(value, '$.repoId'), json_extract(value, '$.legacyNumber'), lower(json_extract(value, '$.workspace')),
55 json_extract(value, '$.number'), json_extract(value, '$.itemId'), ?1
56 FROM json_each(?2)",
57 // Threads the work item's own events started before it had its number.
58 "UPDATE inbox_items SET number = m.number, workspace = m.workspace
59 FROM work_migrated m
60 WHERE m.event_id = ?1 AND inbox_items.thread = 'item:' || m.item_id AND inbox_items.number IS NOT m.number",
61 // Someone with both threads: the old one's history joins the new one.
62 "UPDATE inbox_activity SET item_id = n.id
63 FROM work_migrated m, inbox_items o, inbox_items n
64 WHERE m.event_id = ?1 AND o.thread = m.repo_id || '#' || m.legacy_number
65 AND n.username = o.username AND n.thread = 'item:' || m.item_id AND inbox_activity.item_id = o.id",
66 "UPDATE inbox_items AS n SET activity = n.activity + o.activity, created_at = min(n.created_at, o.created_at),
67 saved = max(n.saved, o.saved)
68 FROM work_migrated m, inbox_items o
69 WHERE m.event_id = ?1 AND o.thread = m.repo_id || '#' || m.legacy_number
70 AND n.username = o.username AND n.thread = 'item:' || m.item_id",
71 "DELETE FROM inbox_items WHERE id IN (
72 SELECT o.id FROM work_migrated m
73 JOIN inbox_items o ON o.thread = m.repo_id || '#' || m.legacy_number
74 JOIN inbox_items n ON n.username = o.username AND n.thread = 'item:' || m.item_id
75 WHERE m.event_id = ?1)",
76 // The rest of the old threads become the work item's.
77 "UPDATE inbox_items SET thread = 'item:' || m.item_id, subject = 'item', number = m.number, workspace = m.workspace
78 FROM work_migrated m
79 WHERE m.event_id = ?1 AND inbox_items.thread = m.repo_id || '#' || m.legacy_number",
80 // Subscriptions: one chosen on the work item already stands.
81 "DELETE FROM inbox_subscriptions WHERE (thread, username) IN (
82 SELECT o.thread, o.username FROM work_migrated m
83 JOIN inbox_subscriptions o ON o.thread = m.repo_id || '#' || m.legacy_number
84 JOIN inbox_subscriptions n ON n.username = o.username AND n.thread = 'item:' || m.item_id
85 WHERE m.event_id = ?1)",
86 "UPDATE inbox_subscriptions SET thread = 'item:' || m.item_id
87 FROM work_migrated m
88 WHERE m.event_id = ?1 AND inbox_subscriptions.thread = m.repo_id || '#' || m.legacy_number",
89 // The log: each of the page's issues' events, found by its repository
90 // and old number through an index of only such events (migration
91 // 0015), so a page reads what it rewrites and no more of the log.
92 concat!(
93 "UPDATE events SET type = 'work_item.' || substr(type, 7),
94 data = json_set(json_remove(data, '$.issueId', '$.repoId'),
95 '$.itemId', (SELECT m.item_id FROM work_migrated m WHERE m.repo_id = events.repo_id AND m.legacy_number = json_extract(events.data, '$.number')),
96 '$.workspace', (SELECT m.workspace FROM work_migrated m WHERE m.repo_id = events.repo_id AND m.legacy_number = json_extract(events.data, '$.number')),
97 '$.repoIds', json_array(events.repo_id),
98 '$.number', (SELECT m.number FROM work_migrated m WHERE m.repo_id = events.repo_id AND m.legacy_number = json_extract(events.data, '$.number')))
99 WHERE id IN (",
100 page_events_by_number!(),
101 ")
102 AND type IN (SELECT value FROM json_each(?3))"
103 ),
104 // Comments on an issue: on no pull request, and naming no work item yet.
105 concat!(
106 "UPDATE events SET
107 data = json_set(data,
108 '$.itemId', (SELECT m.item_id FROM work_migrated m WHERE m.repo_id = events.repo_id AND m.legacy_number = json_extract(events.data, '$.number')),
109 '$.workspace', (SELECT m.workspace FROM work_migrated m WHERE m.repo_id = events.repo_id AND m.legacy_number = json_extract(events.data, '$.number')),
110 '$.repoIds', json_array(events.repo_id),
111 '$.number', (SELECT m.number FROM work_migrated m WHERE m.repo_id = events.repo_id AND m.legacy_number = json_extract(events.data, '$.number')))
112 WHERE id IN (",
113 page_events_by_number!(),
114 ")
115 AND type IN ('comment.created', 'comment.edited', 'comment.deleted')
116 AND json_extract(data, '$.pullId') IS NULL AND json_extract(data, '$.itemId') IS NULL"
117 ),
118 // What names an issue it is about, or overlaps with: `issue`, `otherIssue`.
119 "UPDATE events SET
120 data = json_set(json_remove(data, '$.issue'),
121 '$.item', (SELECT m.number FROM work_migrated m WHERE m.repo_id = events.repo_id AND m.legacy_number = json_extract(events.data, '$.issue')))
122 WHERE id IN (
123 SELECT e.id FROM work_migrated m JOIN events e INDEXED BY events_issue
124 ON e.repo_id = m.repo_id AND json_extract(e.data, '$.issue') = m.legacy_number AND json_extract(e.data, '$.issue') IS NOT NULL
125 WHERE m.event_id = ?1)",
126 "UPDATE events SET
127 data = json_set(json_remove(data, '$.otherIssue'),
128 '$.otherItem', (SELECT m.number FROM work_migrated m WHERE m.repo_id = events.repo_id AND m.legacy_number = json_extract(events.data, '$.otherIssue')))
129 WHERE id IN (
130 SELECT e.id FROM work_migrated m JOIN events e INDEXED BY events_other_issue
131 ON e.repo_id = m.repo_id AND json_extract(e.data, '$.otherIssue') = m.legacy_number AND json_extract(e.data, '$.otherIssue') IS NOT NULL
132 WHERE m.event_id = ?1)",
133];
134
135/// How many of `?1`, `?2`, `?3` a statement reads: D1 takes no more values
136/// than it has.
137fn bound(sql: &str) -> usize {
138 (1..=3).rev().find(|n| sql.contains(&format!("?{n}"))).unwrap_or(0)
139}
140
141/// The statements a `work.migrated` event from the work service calls for;
142/// none for any other event.
143pub fn moved(db: &D1Database, event: &Event) -> Result<Vec<D1PreparedStatement>> {
144 if event.kind != "work.migrated" || event.source != "work" {
145 return Ok(Vec::new());
146 }
147 let Ok(page) = serde_json::from_value::<WorkMigrated>(event.data.clone()) else {
148 worker::console_error!("work.migrated {} could not be read", event.id);
149 return Ok(Vec::new());
150 };
151 if event.data["truncated"].as_bool() == Some(true) {
152 worker::console_error!("work.migrated {} (page {}) was cut short: some issue numbers are left as they were", event.id, page.page);
153 }
154 page_statements(db, &event.id, &page)
155}
156
157/// The statements for one page, recorded as `id`'s.
158fn page_statements(db: &D1Database, id: &str, page: &WorkMigrated) -> Result<Vec<D1PreparedStatement>> {
159 if page.items.is_empty() {
160 return Ok(Vec::new());
161 }
162 let items = serde_json::to_string(&page.items)?;
163 let issue_events = serde_json::to_string(&ISSUE_EVENTS)?;
164 let values: [worker::wasm_bindgen::JsValue; 3] = [id.into(), items.as_str().into(), issue_events.as_str().into()];
165 MOVED.iter().map(|sql| db.prepare(*sql).bind(&values[..bound(sql)])).collect()
166}
167
168/// The cron: pages of `work.migrated` that never arrived, read from the
169/// work service (`g1t_kit::work_migrated`) and applied as the event's
170/// would have been, each recorded as `catch-up:<page>`.
171pub async fn catch_up(db: &D1Database, work: &Fetcher) {
172 let caught = g1t_kit::work_migrated::catch_up(db, work, |page| async move {
173 let statements = page_statements(db, &format!("catch-up:{}", page.page), &page)?;
174 if !statements.is_empty() {
175 db.batch(statements).await?;
176 }
177 Ok(())
178 })
179 .await;
180 if let Err(error) = caught {
181 worker::console_error!("catching up on work.migrated: {error}");
182 }
183}
184
185#[cfg(test)]
186mod tests {
187 use super::*;
188 use crate::schema::{database_before, migrate_from};
189 use rusqlite::{Connection, params};
190
191 const ITEM: &str = "INSERT INTO inbox_items (id, username, event_id, reason, severity, title, body, workspace, repo_id, repo,
192 subject, number, created_at, thread, updated_at, bumped, activity, read_at, saved)
193 VALUES (?1, ?2, ?3, 'author', 'info', 't', 'b', 'acme', ?4, 'acme/rocket', ?5, ?6, ?7, ?8, ?7, ?1, ?9, ?10, ?11)";
194 const EVENT: &str = "INSERT INTO events (id, type, source, time, repo_id, data, seq) VALUES (?1, ?2, 'work', 't', ?3, ?4, ?5)";
195
196 /// One page: rep_1's issue 3 is acme#40, its issue 5 acme#41.
197 fn page() -> String {
198 serde_json::json!([
199 { "repoId": "rep_1", "legacyNumber": 3, "workspace": "Acme", "number": 40, "itemId": "iss_3" },
200 { "repoId": "rep_1", "legacyNumber": 5, "workspace": "Acme", "number": 41, "itemId": "iss_5" },
201 ])
202 .to_string()
203 }
204
205 fn run(db: &Connection, event_id: &str, items: &str) {
206 let issue_events = serde_json::to_string(&ISSUE_EVENTS).unwrap();
207 let values = [event_id, items, issue_events.as_str()];
208 for sql in MOVED {
209 db.execute(sql, rusqlite::params_from_iter(&values[..bound(sql)])).unwrap_or_else(|error| panic!("{sql}: {error}"));
210 }
211 }
212
213 fn logged(db: &Connection, id: &str) -> (String, serde_json::Value) {
214 let (kind, data): (String, String) =
215 db.query_row("SELECT type, data FROM events WHERE id = ?1", [id], |row| Ok((row.get(0)?, row.get(1)?))).unwrap();
216 (kind, serde_json::from_str(&data).unwrap())
217 }
218
219 type ItemState = (String, String, String, Option<u32>, String, u32, Option<String>);
220
221 fn items(db: &Connection) -> Vec<ItemState> {
222 let mut statement = db
223 .prepare("SELECT id, username, thread, number, workspace, activity, read_at FROM inbox_items ORDER BY id")
224 .unwrap();
225 statement
226 .query_map([], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?, row.get(4)?, row.get(5)?, row.get(6)?)))
227 .unwrap()
228 .map(Result::unwrap)
229 .collect()
230 }
231
232 #[test]
233 fn a_page_reads_only_its_own_issues_events() {
234 let db = database_before("9999");
235 for sql in MOVED.iter().filter(|sql| sql.contains("UPDATE events")) {
236 let mut statement = db.prepare(&format!("EXPLAIN QUERY PLAN {sql}")).unwrap();
237 let plan: Vec<String> = statement
238 .query_map(rusqlite::params_from_iter(["evt_9", "[]", "[]"].iter().take(bound(sql))), |row| row.get::<_, String>(3))
239 .unwrap()
240 .map(Result::unwrap)
241 .collect();
242 // Never the whole log, nor every event of the page's repositories.
243 assert!(
244 plan.iter().all(|step| !step.starts_with("SCAN events") && !step.starts_with("SCAN e") && !step.contains("INDEX events_repo")),
245 "{sql}\n{plan:#?}"
246 );
247 assert!(plan.iter().any(|step| step.contains("INDEX events_issue") || step.contains("INDEX events_other_issue")), "{sql}\n{plan:#?}");
248 }
249 }
250
251 #[test]
252 fn what_was_kept_by_issue_number_follows_the_work_item() {
253 let db = database_before("0013_work_items.sql");
254 // ana: a thread on issue 3, and one on pull request 4.
255 db.execute(ITEM, params!["ntf_1", "ana", "evt_1", "rep_1", "issue", 3, "2026-01-01", "rep_1#3", 2, None::<String>, 0]).unwrap();
256 db.execute(ITEM, params!["ntf_2", "ana", "evt_2", "rep_1", "pull", 4, "2026-01-02", "rep_1#4", 1, None::<String>, 0]).unwrap();
257 // bo: a thread on issue 5, and the work item's own since, read, before it had its number.
258 db.execute(ITEM, params!["ntf_3", "bo", "evt_3", "rep_1", "issue", 5, "2026-01-01", "rep_1#5", 3, None::<String>, 1]).unwrap();
259 db.execute(ITEM, params!["ntf_4", "bo", "evt_4", "rep_1", "item", None::<u32>, "2026-02-01", "item:iss_5", 1, "2026-02-02", 0])
260 .unwrap();
261 for (id, item, event) in [("act_1", "ntf_3", "evt_3"), ("act_2", "ntf_4", "evt_4")] {
262 db.execute(
263 "INSERT INTO inbox_activity (id, item_id, username, event_id, reason, severity, title, body, created_at)
264 VALUES (?1, ?2, 'bo', ?3, 'author', 'info', 't', 'b', 't')",
265 params![id, item, event],
266 )
267 .unwrap();
268 }
269 for (user, thread, state) in
270 [("ana", "rep_1#3", "subscribed"), ("ana", "rep_1#4", "ignored"), ("bo", "rep_1#5", "ignored"), ("bo", "item:iss_5", "subscribed")]
271 {
272 db.execute(
273 "INSERT INTO inbox_subscriptions (username, thread, repo_id, state, created_at) VALUES (?1, ?2, 'rep_1', ?3, 't')",
274 params![user, thread, state],
275 )
276 .unwrap();
277 }
278 db.execute(
279 "INSERT INTO inbox_watching (username, repo_id, level, events, updated_at)
280 VALUES ('cy', 'rep_1', 'custom', '[\"issues\",\"deployments\"]', 't')",
281 [],
282 )
283 .unwrap();
284 let events = [
285 ("evt_1", "issue.opened", r#"{"issueId":"iss_3","repoId":"rep_1","number":3}"#),
286 ("evt_2", "pull.opened", r#"{"repoId":"rep_1","number":4,"issue":3}"#),
287 ("evt_3", "comment.created", r#"{"commentId":"cmt_1","repoId":"rep_1","number":5}"#),
288 ("evt_5", "comment.created", r#"{"commentId":"cmt_2","repoId":"rep_1","number":4,"pullId":"pr_4"}"#),
289 ("evt_6", "work.overlap", r#"{"issue":5,"otherIssue":3}"#),
290 ("evt_7", "issue.closed", r#"{"issueId":"iss_9","repoId":"rep_1","number":9}"#),
291 ];
292 for (seq, (id, kind, json)) in events.iter().enumerate() {
293 db.execute(EVENT, params![id, kind, "rep_1", json, seq as i64 + 1]).unwrap();
294 }
295
296 migrate_from(&db, "0013_work_items.sql");
297 // What needs only names is done at once.
298 let subject: String = db.query_row("SELECT subject FROM inbox_items WHERE id = 'ntf_1'", [], |row| row.get(0)).unwrap();
299 assert_eq!(subject, "item");
300 let watched: String = db.query_row("SELECT events FROM inbox_watching WHERE username = 'cy'", [], |row| row.get(0)).unwrap();
301 assert_eq!(watched, r#"["work_items","deployments"]"#);
302 // A subscription to a work item with no origin has no repository.
303 db.execute(
304 "INSERT INTO inbox_subscriptions (username, thread, repo_id, state, created_at) VALUES ('dee', 'item:wki_9', NULL, 'subscribed', 't')",
305 [],
306 )
307 .unwrap();
308
309 // The page, twice: the second time changes nothing.
310 run(&db, "evt_9", &page());
311 let first = items(&db);
312 run(&db, "evt_9", &page());
313 assert_eq!(items(&db), first);
314 // And again as the cron's catch-up reads it: nothing more either.
315 let log = |db: &Connection| -> Vec<(String, serde_json::Value)> {
316 ["evt_1", "evt_2", "evt_3", "evt_5", "evt_6", "evt_7"].iter().map(|id| logged(db, id)).collect()
317 };
318 let logged_once = log(&db);
319 run(&db, "catch-up:1", &page());
320 assert_eq!(items(&db), first);
321 assert_eq!(log(&db), logged_once);
322 assert_eq!(
323 first,
324 vec![
325 ("ntf_1".into(), "ana".into(), "item:iss_3".into(), Some(40), "acme".into(), 2, None),
326 ("ntf_2".into(), "ana".into(), "rep_1#4".into(), Some(4), "acme".into(), 1, None),
327 // bo's old thread joined the work item's, with its history.
328 ("ntf_4".into(), "bo".into(), "item:iss_5".into(), Some(41), "acme".into(), 4, Some("2026-02-02".into())),
329 ]
330 );
331 let moved: u32 = db.query_row("SELECT count(*) FROM inbox_activity WHERE item_id = 'ntf_4'", [], |row| row.get(0)).unwrap();
332 assert_eq!(moved, 2);
333 let since: String = db.query_row("SELECT created_at FROM inbox_items WHERE id = 'ntf_4'", [], |row| row.get(0)).unwrap();
334 assert_eq!(since, "2026-01-01");
335
336 let mut statement = db.prepare("SELECT username, thread, state FROM inbox_subscriptions ORDER BY username, thread").unwrap();
337 let subscriptions: Vec<(String, String, String)> =
338 statement.query_map([], |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?))).unwrap().map(Result::unwrap).collect();
339 assert_eq!(
340 subscriptions,
341 vec![
342 ("ana".into(), "item:iss_3".into(), "subscribed".into()),
343 ("ana".into(), "rep_1#4".into(), "ignored".into()),
344 // What bo chose on the work item stands.
345 ("bo".into(), "item:iss_5".into(), "subscribed".into()),
346 ("dee".into(), "item:wki_9".into(), "subscribed".into()),
347 ]
348 );
349
350 // The log names the work items.
351 let (kind, opened) = logged(&db, "evt_1");
352 assert_eq!(kind, "work_item.opened");
353 assert_eq!(opened, serde_json::json!({ "itemId": "iss_3", "workspace": "acme", "number": 40, "repoIds": ["rep_1"] }));
354 assert_eq!(logged(&db, "evt_2").1, serde_json::json!({ "repoId": "rep_1", "number": 4, "item": 40 }));
355 let comment = logged(&db, "evt_3").1;
356 assert_eq!((comment["number"].as_u64(), comment["itemId"].as_str()), (Some(41), Some("iss_5")));
357 assert_eq!(logged(&db, "evt_6").1, serde_json::json!({ "item": 41, "otherItem": 40 }));
358 // A pull request's comment, and an issue no page has named yet, stay.
359 assert_eq!(logged(&db, "evt_5").1["number"], 4);
360 assert_eq!(logged(&db, "evt_7").0, "issue.closed");
361 }
362}