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