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

Commit

Events: work items in the inbox, the log and live pages

A work item is one inbox thread (item:<id>) in its workspace, linked at /<workspace>/-/work/<n>, read from the work service by its id. Watchers of every repository it is linked to hear of it; it is kept with its origin, shown to the workspace's members and whoever can read the origin, and pushed live only to those who may see it. Watches choose work_items, the digest counts work items, the log's numbers filter reads item, and live pages of every linked repository refresh. Migration 0013 renames what needs only names, lets subscriptions keep no repository, and records work.migrated pages, from which threads, subscriptions and the log's issue events are rewritten to the work items, idempotently.

G1T-Session:
ses_01m566c3bte71t8fa4g03g1c8v
syntaqxcommitted Parentab7b90cBrowse files
10 files+987−2350/10 viewed
+1−0
15031503 dependencies = [
15041504 "g1t-contracts",
15051505 "g1t-kit",
1506+ "rusqlite",
15061507 "serde",
15071508 "serde_json",
15081509 "worker",
+3−0
1515 serde_json.workspace = true
1616 worker.workspace = true
1717
18+[dev-dependencies]
19+rusqlite = { version = "0.32", features = ["bundled"] }
20+
1821 # wasm-opt at -O1: about the same gzipped size as -O in a tenth of the
1922 # time.
2023 [package.metadata.wasm-pack.profile.release]
+55−0
1+-- Issues became work items (g1t #57). What needs only names is rewritten
2+-- here; what was kept by a repository's issue number (inbox threads,
3+-- subscriptions, the log's events) is rewritten as the work service
4+-- announces where each issue went (`work.migrated`, src/work_items.rs),
5+-- recorded in `work_migrated`.
6+
7+-- Inbox items about an issue are about its work item.
8+UPDATE inbox_items SET subject = 'item' WHERE subject = 'issue';
9+
10+-- A custom watch that followed issues follows work items.
11+UPDATE inbox_watching
12+SET events = (
13+ SELECT json_group_array(CASE value WHEN 'issues' THEN 'work_items' ELSE value END)
14+ FROM json_each(inbox_watching.events)
15+)
16+WHERE EXISTS (SELECT 1 FROM json_each(inbox_watching.events) WHERE value = 'issues');
17+
18+-- Subscriptions to a work item keep its origin as `repo_id`, and one
19+-- with no origin has none.
20+CREATE TABLE inbox_subscriptions_new (
21+ username TEXT NOT NULL,
22+ -- `<repo_id>#<number>` for a pull request, `item:<item_id>` for a work item.
23+ thread TEXT NOT NULL,
24+ -- The pull request's repository, or the work item's origin.
25+ repo_id TEXT,
26+ -- subscribed | unsubscribed | ignored
27+ state TEXT NOT NULL,
28+ -- Why: comment, mention, assign, review_requested, author or manual.
29+ reason TEXT,
30+ created_at TEXT NOT NULL,
31+ -- When the person last chose (subscribed, unsubscribed or ignored by
32+ -- hand); null while it only follows what they did.
33+ chosen_at TEXT,
34+ PRIMARY KEY (thread, username)
35+);
36+INSERT INTO inbox_subscriptions_new (username, thread, repo_id, state, reason, created_at, chosen_at)
37+SELECT username, thread, repo_id, state, reason, created_at, chosen_at FROM inbox_subscriptions;
38+DROP TABLE inbox_subscriptions;
39+ALTER TABLE inbox_subscriptions_new RENAME TO inbox_subscriptions;
40+CREATE INDEX inbox_subscriptions_repo ON inbox_subscriptions (repo_id);
41+
42+-- Where each repository's issues went, a page of `work.migrated` at a time.
43+CREATE TABLE work_migrated (
44+ repo_id TEXT NOT NULL,
45+ -- The issue's number in its repository.
46+ legacy_number INTEGER NOT NULL,
47+ -- The work item's workspace (a slug), number in it, and id.
48+ workspace TEXT NOT NULL,
49+ number INTEGER NOT NULL,
50+ item_id TEXT NOT NULL,
51+ -- The `work.migrated` event that said so.
52+ event_id TEXT NOT NULL,
53+ PRIMARY KEY (repo_id, legacy_number)
54+);
55+CREATE INDEX work_migrated_event ON work_migrated (event_id);
+16−15
1717
1818 use g1t_contracts::events::{
1919 ActivityDigest, ActivityDigestArgs, ActorCount, DefaultBranchPushes, DeploymentDigest, Event,
20− FolioDigest, FolioEdited, IssueDigest, MAX_DIGEST_EVENTS, MAX_DIGEST_LISTED, MAX_DIGEST_REPOS,
20+ FolioDigest, FolioEdited, ItemDigest, MAX_DIGEST_EVENTS, MAX_DIGEST_LISTED, MAX_DIGEST_REPOS,
2121 PackageDigest, PullDigest, PushDigest, PushesBy, ReleaseDigest, RepoDigest,
2222 };
2323 use g1t_contracts::id_floor;
2727
2828 use crate::EventRow;
2929
30−/// The repository events a digest counts.
30+/// The repository events a digest counts. A work item's count in its
31+/// origin's, the repository its events carry; one with no origin in none.
3132 const REPO_TYPES: [&str; 12] = [
3233 "git.push",
3334 "pull.opened",
3435 "pull.merged",
3536 "pull.closed",
36− "issue.opened",
37− "issue.closed",
37+ "work_item.opened",
38+ "work_item.closed",
3839 "review.completed",
3940 "comment.created",
4041 "deployment.succeeded",
237238 pulls_opened: Tally,
238239 pulls_merged: Tally,
239240 pulls_closed: Tally,
240− issues_opened: Tally,
241− issues_closed: Tally,
241+ items_opened: Tally,
242+ items_closed: Tally,
242243 reviews: Tally,
243244 comments: Tally,
244245 deploys_succeeded: Tally,
279280 "pull.opened" => self.pulls_opened.add(actor),
280281 "pull.merged" => self.pulls_merged.add(actor),
281282 "pull.closed" => self.pulls_closed.add(actor),
282− "issue.opened" => self.issues_opened.add(actor),
283− "issue.closed" => self.issues_closed.add(actor),
283+ "work_item.opened" => self.items_opened.add(actor),
284+ "work_item.closed" => self.items_closed.add(actor),
284285 "review.completed" => self.reviews.add(actor),
285286 "comment.created" => {
286287 // One of the workspace's agents, as itself: the event's actor
356357 merged: self.pulls_merged.into_counts(),
357358 closed: self.pulls_closed.into_counts(),
358359 },
359− issues: IssueDigest {
360− opened: self.issues_opened.into_counts(),
361− closed: self.issues_closed.into_counts(),
360+ items: ItemDigest {
361+ opened: self.items_opened.into_counts(),
362+ closed: self.items_closed.into_counts(),
362363 },
363364 reviews: self.reviews.into_counts(),
364365 comments: self.comments.into_counts(),
582583 ),
583584 event(
584585 "evt_5",
585− "issue.opened",
586+ "work_item.opened",
586587 Some("rep_2"),
587588 Some("usr_bo"),
588589 json!({ "number": 4 }),
589590 ),
590591 event(
591592 "evt_6",
592− "issue.closed",
593+ "work_item.closed",
593594 Some("rep_2"),
594595 g1t,
595596 json!({ "number": 4 }),
755756 }]
756757 );
757758 assert_eq!(
758− api.issues.opened,
759+ api.items.opened,
759760 vec![ActorCount {
760761 actor: "user:usr_bo".into(),
761762 count: 1
762763 }]
763764 );
764765 assert_eq!(
765− api.issues.closed,
766+ api.items.closed,
766767 vec![ActorCount {
767768 actor: "user:usr_g1t_agent".into(),
768769 count: 1
+426−159
A large diff (1,103 lines) is not shown until you ask for it.
+10−7
2626 mod fanout;
2727 mod inbox;
2828 mod live;
29+#[cfg(test)]
30+mod schema;
2931 mod security_digest;
3032 mod subscriptions;
33+mod work_items;
3134
3235 use g1t_contracts::events::{Event, ListArgs, MAX_LIST_REPOS, PublishArgs};
3336 use g1t_contracts::new_id;
497500 values.push(Bind::Text(actor.clone()));
498501 }
499502 if !a.numbers.is_empty() {
500− // An issue or pull request's own events name it as `number`; a
501− // comment, review or link on it names it as `issue`.
503+ // A work item's or pull request's own events name it as `number`; a
504+ // comment, review or link on it names it as `item`.
502505 let marks = vec!["?"; a.numbers.len()].join(", ");
503506 conditions.push(format!(
504− "(json_extract(data, '$.number') IN ({marks}) OR json_extract(data, '$.issue') IN ({marks}))"
507+ "(json_extract(data, '$.number') IN ({marks}) OR json_extract(data, '$.item') IN ({marks}))"
505508 ));
506509 for _ in 0..2 {
507510 values.extend(a.numbers.iter().map(|number| Bind::Number(f64::from(*number))));
564567 fn numbers_match_an_item_or_what_is_said_on_it_since_a_time() {
565568 let (filter, binds) = list_filter(&ListArgs {
566569 repo_id: Some("rep_1".into()),
567− types: vec!["issue.opened".into()],
570+ types: vec!["work_item.opened".into()],
568571 numbers: vec![4, 9],
569572 since: Some("2026-10-01T00:00:00Z".into()),
570573 actor: Some("usr_g1t_agent".into()),
573576 assert_eq!(
574577 filter,
575578 "WHERE repo_id = ? AND type IN (?) AND actor = ? AND \
576− (json_extract(data, '$.number') IN (?, ?) OR json_extract(data, '$.issue') IN (?, ?)) AND time >= ?"
579+ (json_extract(data, '$.number') IN (?, ?) OR json_extract(data, '$.item') IN (?, ?)) AND time >= ?"
577580 );
578581 assert_eq!(
579582 binds,
580583 vec![
581584 Bind::Text("rep_1".into()),
582− Bind::Text("issue.opened".into()),
585+ Bind::Text("work_item.opened".into()),
583586 Bind::Text("usr_g1t_agent".into()),
584587 Bind::Number(4.0),
585588 Bind::Number(9.0),
604607 filter,
605608 format!(
606609 "WHERE repo_id = ? AND type IN (?, ?) AND \
607− (json_extract(data, '$.number') IN (?) OR json_extract(data, '$.issue') IN (?)) AND time >= ? AND {AFTER}"
610+ (json_extract(data, '$.number') IN (?) OR json_extract(data, '$.item') IN (?)) AND time >= ? AND {AFTER}"
608611 )
609612 );
610613 // The cursor's id is bound once to find its place, and once for a
+55−1
44 //! (more for a very busy one); nothing written in the repository goes
55 //! along, and who may see it is notify's to keep: a tab watches a
66 //! repository only after the site checked its person may read it.
7+//!
8+//! A work item's events and comments go to every repository it is linked
9+//! to (`repoIds`), not only its origin, which it may not have.
710
811 use g1t_contracts::events::Event;
9−use g1t_contracts::live::{REPO_CHANGED, RepoChangedArgs, changes_of};
12+use g1t_contracts::live::{MAX_CHANGES, REPO_CHANGED, RepoChange, RepoChangedArgs, change_of};
1013 use worker::{Fetcher, Result};
1114
15+/// The changes an event makes: one for its repository, and one for each
16+/// other repository it names in `repoIds`.
17+pub(crate) fn changes_for(event: &Event) -> Vec<RepoChange> {
18+ let mut changes: Vec<RepoChange> = change_of(event).into_iter().collect();
19+ for repo_id in event.data["repoIds"].as_array().into_iter().flatten().filter_map(|id| id.as_str()) {
20+ if repo_id.is_empty() || changes.iter().any(|change| change.repo_id == repo_id) {
21+ continue;
22+ }
23+ let linked = Event { repo_id: Some(repo_id.to_owned()), ..event.clone() };
24+ changes.extend(change_of(&linked));
25+ }
26+ changes
27+}
28+
29+/// A batch's changes, in order, in calls of at most [`MAX_CHANGES`].
30+pub(crate) fn changes_of(events: &[Event]) -> Vec<Vec<RepoChange>> {
31+ let changes: Vec<RepoChange> = events.iter().flat_map(changes_for).collect();
32+ changes.chunks(MAX_CHANGES).map(<[RepoChange]>::to_vec).collect()
33+}
34+
1235 /// Tells notify of the batch's changes. Logged and left when it cannot be
1336 /// reached: pages then catch up by themselves, and nothing else waits on it.
1437 pub(crate) async fn tell(notify: &Fetcher, events: &[Event]) {
1942 }
2043 }
2144 }
45+
46+#[cfg(test)]
47+mod tests {
48+ use super::*;
49+ use serde_json::json;
50+
51+ fn event(kind: &str, repo_id: Option<&str>, data: serde_json::Value) -> Event {
52+ Event {
53+ id: "evt_1".into(),
54+ kind: kind.into(),
55+ source: "work".into(),
56+ time: "2026-10-11T10:00:00Z".into(),
57+ repo_id: repo_id.map(Into::into),
58+ actor: None,
59+ data,
60+ }
61+ }
62+
63+ #[test]
64+ fn a_work_item_changes_every_repository_it_is_linked_to() {
65+ let data = json!({ "itemId": "wki_1", "workspace": "acme", "number": 40, "repoIds": ["rep_1", "rep_2"] });
66+ let repos = |event: &Event| changes_for(event).into_iter().map(|change| (change.repo_id, change.number)).collect::<Vec<_>>();
67+ let with_origin = event("work_item.updated", Some("rep_1"), data.clone());
68+ assert_eq!(repos(&with_origin), vec![("rep_1".to_owned(), Some(40)), ("rep_2".to_owned(), Some(40))]);
69+ // Filed in the workspace: only the repositories it is linked to.
70+ let without = event("work_item.linked", None, data);
71+ assert_eq!(repos(&without).len(), 2);
72+ // What changes no page stays quiet.
73+ assert!(changes_for(&event("audit.recorded", None, json!({ "repoIds": ["rep_1"] }))).is_empty());
74+ }
75+}
+44−0
1+//! The events database as the migrations make it, for tests: SQLite (which
2+//! D1 is), each migration in a transaction of its own as D1 applies it.
3+
4+use rusqlite::Connection;
5+
6+/// The migrations, in order: each file's name and SQL.
7+pub(crate) fn migrations() -> Vec<(String, String)> {
8+ let dir = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("migrations");
9+ let mut names: Vec<String> = std::fs::read_dir(&dir)
10+ .unwrap()
11+ .map(|entry| entry.unwrap().file_name().into_string().unwrap())
12+ .filter(|name| name.ends_with(".sql"))
13+ .collect();
14+ names.sort();
15+ names.into_iter().map(|name| (name.clone(), std::fs::read_to_string(dir.join(&name)).unwrap())).collect()
16+}
17+
18+/// Applies one migration as D1 does: in one transaction.
19+pub(crate) fn apply(db: &Connection, name: &str, sql: &str) {
20+ db.execute_batch(&format!("BEGIN;\n{sql}\nCOMMIT;")).unwrap_or_else(|error| panic!("{name}: {error}"));
21+}
22+
23+/// An in-memory database with the migrations before `name` applied, for
24+/// filling with what that migration then moves.
25+pub(crate) fn database_before(name: &str) -> Connection {
26+ let db = Connection::open_in_memory().unwrap();
27+ db.execute_batch("PRAGMA foreign_keys = ON;").unwrap();
28+ for (file, sql) in migrations() {
29+ if file.as_str() >= name {
30+ break;
31+ }
32+ apply(&db, &file, &sql);
33+ }
34+ db
35+}
36+
37+/// Applies the migrations from `name` on.
38+pub(crate) fn migrate_from(db: &Connection, name: &str) {
39+ for (file, sql) in migrations() {
40+ if file.as_str() >= name {
41+ apply(db, &file, &sql);
42+ }
43+ }
44+}
+87−53
1−//! Who follows what, for the inbox (inbox.rs): subscriptions to issues and
1+//! Who follows what, for the inbox (inbox.rs): subscriptions to work items and
22 //! pull requests, how people watch repositories, and what each person
33 //! chose about being told. See `g1t_contracts::inbox`.
44 //!
5−//! An issue's or pull request's author (or whoever asked g1t for it), its
5+//! A work item's or pull request's author (or whoever asked g1t for it), its
66 //! assignees and its reviewers are subscribed without a row. A row records
77 //! the rest: someone who commented or was mentioned (subscribed as they
88 //! did), or who subscribed, unsubscribed or ignored it by hand. Commenting
99 //! or being mentioned subscribes again someone who unsubscribed; nothing
1010 //! but their own choice undoes ignoring.
1111
12−use std::collections::HashMap;
12+use std::collections::{HashMap, HashSet};
1313
1414 use g1t_contracts::events::Event;
1515 use g1t_contracts::inbox::*;
1919 use worker::wasm_bindgen::JsValue;
2020 use worker::{D1Database, D1PreparedStatement, Fetcher, Result};
2121
22−use crate::inbox::{is_g1t, is_g1t_id, numbered_thread};
22+use crate::inbox::{ITEM_THREAD, is_g1t, is_g1t_id, item_thread, numbered_thread, subscribable};
2323
2424 /// The most watchers read for one repository's event.
2525 const MAX_WATCHERS: u32 = 5000;
9797 .map(|row| (row.username.clone(), row.reason.unwrap_or(Reason::Manual)))
9898 }
9999
100− /// Everyone watching the repository for `kind` of activity: `issues`,
100+ /// Everyone watching the repository for `kind` of activity: `work_items`,
101101 /// `pulls`, `deployments` or `security`.
102102 pub fn watching<'a>(&'a self, kind: &'a str) -> impl Iterator<Item = String> + 'a {
103103 self.watchers
129129 }
130130 }
131131
132−/// An issue's or pull request's subscription rows.
132+/// A work item's or pull request's subscription rows.
133133 pub async fn of_thread(db: &D1Database, thread: &str) -> Result<Vec<Subscription>> {
134134 Ok(db
135135 .prepare("SELECT username, state, reason, chosen_at FROM inbox_subscriptions WHERE thread = ?")
186186 .collect())
187187 }
188188
189−/// Subscribes the people an event names to its issue or pull request:
189+/// The watchers of a work item: its origin's (`own`) as they watch it,
190+/// then those of the other repositories it is linked to who do not watch
191+/// the origin, each once; ignoring another repository says nothing about
192+/// it. Also who was found only through another repository.
193+pub fn linked_watchers(own: Vec<Watcher>, others: &[Vec<Watcher>]) -> (Vec<Watcher>, HashSet<String>) {
194+ let mut watchers = own;
195+ let mut elsewhere = HashSet::new();
196+ for watcher in others.iter().flatten() {
197+ if watcher.level == WatchLevel::Ignore || watchers.iter().any(|had| had.username.eq_ignore_ascii_case(&watcher.username)) {
198+ continue;
199+ }
200+ elsewhere.insert(watcher.username.to_lowercase());
201+ watchers.push(watcher.clone());
202+ }
203+ (watchers, elsewhere)
204+}
205+
206+/// Subscribes the people an event names to its work item or pull request:
190207 /// anyone without a row, and anyone who had unsubscribed. Someone who
191−/// ignores it stays ignoring it.
208+/// ignores it stays ignoring it. `repo_id` is the pull request's
209+/// repository, or the work item's origin.
192210 pub fn auto_subscribe(
193211 db: &D1Database,
194212 thread: &str,
195− repo_id: &str,
213+ repo_id: Option<&str>,
196214 people: &[(String, Reason)],
197215 at: &str,
198216 ) -> Result<Vec<D1PreparedStatement>> {
199− if !thread.contains('#') {
217+ if !subscribable(thread) {
200218 return Ok(Vec::new());
201219 }
202220 people
211229 .bind(&[
212230 username.as_str().into(),
213231 thread.into(),
214− repo_id.into(),
232+ repo_id.map_or(JsValue::NULL, JsValue::from),
215233 reason.as_str().into(),
216234 at.into(),
217235 ])
260278 thread: Option<String>,
261279 repo_id: Option<String>,
262280 repo: Option<String>,
281+ workspace: Option<String>,
263282 number: Option<f64>,
264283 subject: Option<String>,
265284 }
266285
267−/// The issue or pull request asked about: its thread, repository and
268−/// number, and its `owner/name` when known.
286+/// The work item or pull request asked about: its thread, its repository
287+/// (a work item's origin), its id when a work item, its number, and where
288+/// it is (`owner/name`, or a work item's workspace) when known.
269289 struct Target {
270290 thread: String,
271− repo_id: String,
291+ repo_id: Option<String>,
292+ item_id: Option<String>,
272293 number: u32,
273294 repo: Option<String>,
274295 }
275296
297+impl Target {
298+ /// What the work service is asked, to know it is there and who takes part.
299+ fn subject_args(&self) -> InboxSubjectArgs {
300+ InboxSubjectArgs {
301+ repo_id: self.repo_id.clone(),
302+ number: self.number,
303+ item_id: self.item_id.clone(),
304+ comment_id: None,
305+ }
306+ }
307+}
308+
276309 async fn target(db: &D1Database, username: &str, a: &SubscriptionArgs) -> Result<Option<Target>> {
277310 if let Some(id) = &a.id {
278311 let row = db
279− .prepare("SELECT thread, repo_id, repo, number, subject FROM inbox_items WHERE id = ? AND username = ?")
312+ .prepare("SELECT thread, repo_id, repo, workspace, number, subject FROM inbox_items WHERE id = ? AND username = ?")
280313 .bind(&[id.as_str().into(), username.into()])?
281314 .first::<ItemRow>(None)
282315 .await?;
283316 return Ok(row.and_then(|row| {
284− let kind = row.subject.as_deref().and_then(SubjectKind::parse);
285− if !matches!(kind, Some(SubjectKind::Issue | SubjectKind::Pull)) {
286− return None;
317+ let thread = row.thread?;
318+ let number = row.number? as u32;
319+ match row.subject.as_deref().and_then(SubjectKind::parse) {
320+ Some(SubjectKind::Item) => Some(Target {
321+ item_id: Some(thread.strip_prefix(ITEM_THREAD)?.to_owned()),
322+ thread,
323+ repo_id: row.repo_id,
324+ number,
325+ repo: row.workspace,
326+ }),
327+ Some(SubjectKind::Pull) => Some(Target {
328+ thread,
329+ repo_id: Some(row.repo_id?),
330+ item_id: None,
331+ number,
332+ repo: row.repo,
333+ }),
334+ _ => None,
287335 }
288− Some(Target {
289− thread: row.thread?,
290− repo_id: row.repo_id?,
291− number: row.number? as u32,
292− repo: row.repo,
293− })
294336 }));
295337 }
338+ if let (Some(item_id), Some(number)) = (&a.item_id, a.number) {
339+ return Ok(Some(Target {
340+ thread: item_thread(item_id),
341+ repo_id: None,
342+ item_id: Some(item_id.clone()),
343+ number,
344+ repo: None,
345+ }));
346+ }
296347 let (Some(repo_id), Some(number)) = (&a.repo_id, a.number) else {
297348 return Ok(None);
298349 };
299350 Ok(Some(Target {
300351 thread: numbered_thread(repo_id, number),
301− repo_id: repo_id.clone(),
352+ repo_id: Some(repo_id.clone()),
353+ item_id: None,
302354 number,
303355 repo: None,
304356 }))
329381 }
330382 }
331383
332−/// The person's subscription to an issue or pull request. None when there
333−/// is no such issue or pull request, or it is a thread nobody subscribes to.
384+/// The person's subscription to a work item or pull request. None when there
385+/// is no such work item or pull request, or it is a thread nobody subscribes to.
334386 pub async fn subscription(db: &D1Database, work: &Fetcher, a: SubscriptionArgs) -> Result<Option<ThreadSubscription>> {
335387 let Some(viewer) = &a.viewer else {
336388 return Ok(None);
344396 .bind(&[target.thread.as_str().into(), username.as_str().into()])?
345397 .first::<SubscriptionRow>(None)
346398 .await?;
347− // Without a row, whether they take part is the issue's to say; and an
348− // issue that is not there has no subscription.
349− let subject: Option<InboxSubject> = g1t_kit::call(
350− work,
351− "inbox_subject",
352− &InboxSubjectArgs {
353− repo_id: target.repo_id.clone(),
354− number: target.number,
355− comment_id: None,
356− },
357− )
358− .await?;
399+ // Without a row, whether they take part is the work item's or pull
400+ // request's to say; and one that is not there has no subscription.
401+ let subject: Option<InboxSubject> = g1t_kit::call(work, "inbox_subject", &target.subject_args()).await?;
359402 let Some(subject) = subject else {
360403 return Ok(None);
361404 };
383426 let Some(target) = target(db, &username, &a.on).await? else {
384427 return Ok(None);
385428 };
386− // Only an issue or pull request that is there.
387− let found: Option<InboxSubject> = g1t_kit::call(
388− work,
389− "inbox_subject",
390− &InboxSubjectArgs {
391− repo_id: target.repo_id.clone(),
392− number: target.number,
393− comment_id: None,
394− },
395− )
396− .await?;
429+ // Only a work item or pull request that is there.
430+ let found: Option<InboxSubject> = g1t_kit::call(work, "inbox_subject", &target.subject_args()).await?;
397431 if found.is_none() {
398432 return Ok(None);
399433 }
424458 .bind(&[
425459 username.as_str().into(),
426460 target.thread.as_str().into(),
427− target.repo_id.as_str().into(),
461+ target.repo_id.as_deref().map_or(JsValue::NULL, JsValue::from),
428462 state.as_str().into(),
429463 if state == State::Subscribed { Reason::Manual.as_str().into() } else { JsValue::NULL },
430464 now.as_str().into(),
693727 vec![("cy".to_owned(), Reason::Mention), ("dee".to_owned(), Reason::Manual)]
694728 );
695729 assert_eq!(audience.watching("deployments").collect::<Vec<_>>(), vec!["fay", "gus"]);
696− assert_eq!(audience.watching("issues").collect::<Vec<_>>(), vec!["gus"]);
730+ assert_eq!(audience.watching("work_items").collect::<Vec<_>>(), vec!["gus"]);
697731 }
698732
699733 #[test]
700734 fn choices_keep_only_what_is_known() {
701− assert_eq!(custom_events(&["Pulls".into(), "releases".into(), "issues".into(), "pulls".into()]), vec!["issues", "pulls"]);
735+ assert_eq!(custom_events(&["Pulls".into(), "releases".into(), "issues".into(), "work_items".into(), "pulls".into()]), vec!["work_items", "pulls"]);
702736 assert_eq!(email_reasons(&["mention".into(), "nope".into(), "agent".into()]), vec![Reason::Agent, Reason::Mention]);
703737 let settings = SettingsRow { username: "ana".into(), email: None, default_watch: Some("participating".into()), security_alerts: None, security_digest: None }.into_settings();
704738 assert_eq!(settings.email, DEFAULT_EMAIL.to_vec());
+290−0
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+
19+use g1t_contracts::events::{Event, WorkMigrated};
20+use worker::{D1Database, D1PreparedStatement, Result};
21+
22+/// The issue events of the log, by their old names.
23+pub 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.
37+pub 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.
114+fn 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.
120+pub 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)]
141+mod 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+}

Comments

Sign in to comment.