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