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
154 linesCodeBlameRaw
1//! `work.migrated`: where repositories' issues went when they became work
2//! items. Links and deliveries kept by an issue's repository and number
3//! (`legacy_repo_id`, `legacy_number`, from migrations/0013_work_items.sql)
4//! take the work item's id, workspace and number, and lose the old pair.
5//! Each page is applied on its own, and a page seen again changes nothing.
6
7use g1t_contracts::events::{Event, WorkMigrated};
8use worker::{D1Database, Result};
9
10/// The page's moves, from `?1` (its `items` as JSON), as rows.
11const MOVES: &str = "SELECT json_extract(value, '$.repoId') AS repo_id, json_extract(value, '$.legacyNumber') AS legacy_number,
12 json_extract(value, '$.workspace') AS workspace, json_extract(value, '$.number') AS number,
13 json_extract(value, '$.itemId') AS item_id
14 FROM json_each(?1)";
15
16/// What one page rewrites, each taking the page's `items` as `?1`.
17pub fn statements() -> [String; 2] {
18 [
19 format!(
20 "UPDATE links SET item_id = m.item_id, workspace = m.workspace, number = m.number, legacy_repo_id = NULL, legacy_number = NULL
21 FROM ({MOVES}) AS m
22 WHERE links.legacy_repo_id = m.repo_id AND links.legacy_number = m.legacy_number"
23 ),
24 format!(
25 "UPDATE deliveries SET item = m.workspace || '#' || m.number, legacy_repo_id = NULL, legacy_number = NULL
26 FROM ({MOVES}) AS m
27 WHERE deliveries.legacy_repo_id = m.repo_id AND deliveries.legacy_number = m.legacy_number"
28 ),
29 ]
30}
31
32/// Applies `event` when it is `work.migrated`. Whether it was.
33pub async fn on_event(db: &D1Database, event: &Event) -> Result<bool> {
34 if event.kind != "work.migrated" {
35 return Ok(false);
36 }
37 let Ok(page) = serde_json::from_value::<WorkMigrated>(event.data.clone()) else {
38 return Ok(true);
39 };
40 if page.items.is_empty() {
41 return Ok(true);
42 }
43 let items = serde_json::to_string(&page.items)?;
44 let mut batch = Vec::new();
45 for sql in statements() {
46 batch.push(db.prepare(&sql).bind(&[items.as_str().into()])?);
47 }
48 db.batch(batch).await?;
49 Ok(true)
50}
51
52#[cfg(test)]
53mod tests {
54 use rusqlite::{Connection, params};
55
56 use g1t_contracts::work::MigratedItem;
57
58 use super::statements;
59
60 fn migrations() -> Vec<(String, String)> {
61 let dir = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("migrations");
62 let mut names: Vec<String> = std::fs::read_dir(&dir)
63 .unwrap()
64 .map(|entry| entry.unwrap().file_name().into_string().unwrap())
65 .filter(|name| name.ends_with(".sql"))
66 .collect();
67 names.sort();
68 names.into_iter().map(|name| (name.clone(), std::fs::read_to_string(dir.join(&name)).unwrap())).collect()
69 }
70
71 /// Applies the migrations before `name` (or from it on, with `from`),
72 /// each in a transaction of its own as D1 does.
73 fn apply(db: &Connection, name: &str, from: bool) {
74 for (file, sql) in migrations() {
75 if (file.as_str() >= name) == from {
76 db.execute_batch(&format!("BEGIN;\n{sql}\nCOMMIT;")).unwrap_or_else(|error| panic!("{file}: {error}"));
77 }
78 }
79 }
80
81 fn page(items: &[MigratedItem]) -> String {
82 serde_json::to_string(items).unwrap()
83 }
84
85 #[test]
86 fn links_and_deliveries_kept_by_issue_number_move_to_the_work_item() {
87 let db = Connection::open_in_memory().unwrap();
88 apply(&db, "0013", false);
89 db.execute_batch(
90 "INSERT INTO links (id, workspace, connection_id, provider, external_id, key, title, url, repo_id, repo, number,
91 count, announced, told_started, first_seen, last_seen)
92 VALUES ('lnk_1', 'acme', 'con_1', 'sentry', '77', 'WEB-1', 'Boom', 'https://sentry.io/issues/77', 'rep_web', 'acme/web', 4,
93 12, 10, 1, '2026-01-01T00:00:00Z', '2026-01-02T00:00:00Z'),
94 ('lnk_2', 'acme', 'con_2', 'linear', 'uuid', 'ENG-2', 'Do it', 'https://linear.app/acme/issue/ENG-2', 'rep_api', 'acme/api', 4,
95 1, 1, 0, '2026-01-01T00:00:00Z', '2026-01-01T00:00:00Z');
96 INSERT INTO deliveries (id, connection_id, received_at, event, outcome, detail, issue)
97 VALUES ('dlv_1', 'con_1', '2026-01-02T00:00:00Z', 'issue.created', 'opened', 'Opened #4.', 'acme/web#4'),
98 ('dlv_2', 'con_1', '2026-01-02T00:00:00Z', 'request', 'refused', 'Not signed.', NULL);
99 INSERT INTO agent_runs (session_id, workspace, repo, issue, connection_id, actor, provider_session_id, created_at, synced_at)
100 VALUES ('ses_1', 'acme', 'acme/web', 4, 'aip_1', '{}', 'sesn_1', '2026-01-01T00:00:00Z', '2026-01-01T00:00:00Z');",
101 )
102 .unwrap();
103 apply(&db, "0013", true);
104
105 // Until the work service says where the issues went, they keep the old pair.
106 let pending: (Option<String>, Option<u32>, String, u32, u32) = db
107 .query_row("SELECT item_id, number, legacy_repo_id, legacy_number, count FROM links WHERE id = 'lnk_1'", [], |row| {
108 Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?, row.get(4)?))
109 })
110 .unwrap();
111 assert_eq!(pending, (None, None, "rep_web".to_owned(), 4, 12));
112 let delivery: (Option<String>, Option<String>, Option<u32>) = db
113 .query_row("SELECT item, legacy_repo_id, legacy_number FROM deliveries WHERE id = 'dlv_1'", [], |row| {
114 Ok((row.get(0)?, row.get(1)?, row.get(2)?))
115 })
116 .unwrap();
117 assert_eq!(delivery, (None, Some("rep_web".to_owned()), Some(4)));
118 let run: Option<u32> = db.query_row("SELECT item FROM agent_runs WHERE session_id = 'ses_1'", [], |row| row.get(0)).unwrap();
119 assert_eq!(run, None);
120
121 let moved = [
122 MigratedItem { repo_id: "rep_web".into(), legacy_number: 4, workspace: "acme".into(), number: 31, item_id: "iss_web4".into() },
123 MigratedItem { repo_id: "rep_api".into(), legacy_number: 4, workspace: "acme".into(), number: 32, item_id: "iss_api4".into() },
124 MigratedItem { repo_id: "rep_web".into(), legacy_number: 5, workspace: "acme".into(), number: 33, item_id: "iss_web5".into() },
125 ];
126 // Twice: a page seen again changes nothing.
127 for _ in 0..2 {
128 for sql in statements() {
129 db.execute(&sql, params![page(&moved)]).unwrap();
130 }
131 }
132 let link = |id: &str| -> (String, Option<String>, Option<u32>, Option<String>, Option<u32>) {
133 db.query_row("SELECT workspace, item_id, number, legacy_repo_id, legacy_number FROM links WHERE id = ?1", [id], |row| {
134 Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?, row.get(4)?))
135 })
136 .unwrap()
137 };
138 assert_eq!(link("lnk_1"), ("acme".to_owned(), Some("iss_web4".to_owned()), Some(31), None, None));
139 assert_eq!(link("lnk_2"), ("acme".to_owned(), Some("iss_api4".to_owned()), Some(32), None, None));
140 let delivery = |id: &str| -> (Option<String>, Option<String>) {
141 db.query_row("SELECT item, legacy_repo_id FROM deliveries WHERE id = ?1", [id], |row| Ok((row.get(0)?, row.get(1)?))).unwrap()
142 };
143 assert_eq!(delivery("dlv_1"), (Some("acme#31".to_owned()), None));
144 assert_eq!(delivery("dlv_2"), (None, None));
145 let left: u32 = db
146 .query_row(
147 "SELECT (SELECT count(*) FROM links WHERE legacy_repo_id IS NOT NULL) + (SELECT count(*) FROM deliveries WHERE legacy_repo_id IS NOT NULL)",
148 [],
149 |row| row.get(0),
150 )
151 .unwrap();
152 assert_eq!(left, 0);
153 }
154}