| 1 | //! Live repository pages: what the notify service is told when something |
| 2 | //! happens in a repository, so the pages open on it refresh themselves. |
| 3 | //! |
| 4 | //! The events service tells it of every event a repository page shows |
| 5 | //! (`repo_changed`, with [`change_of`] for each), and the actions service |
| 6 | //! of a run's jobs starting and finishing, which are not events. Notify |
| 7 | //! passes each change to the people watching that repository's pages: a |
| 8 | //! tab on one asks to (`watch_repo`, through the site, which checks they |
| 9 | //! may read it first). See docs.g1t.sh/guides/notifications/, "Pages that |
| 10 | //! update live". |
| 11 | //! |
| 12 | //! A change says only what kind of thing moved and where: never a title, a |
| 13 | //! body, a name or anything else written in the repository. |
| 14 | //! |
| 15 | //! Work items live in workspaces, and one may have no repository at all: |
| 16 | //! what happens to a workspace's work items is also told to the people |
| 17 | //! watching its Work pages (`workspace_changed`, with |
| 18 | //! [`workspace_change_of`] for each; a tab asks with `watch_workspace`, |
| 19 | //! through the site, which checks they are a member first). |
| 20 | |
| 21 | use serde::{Deserialize, Serialize}; |
| 22 | |
| 23 | use crate::events::Event; |
| 24 | |
| 25 | /// The notify service's method that takes [`RepoChangedArgs`]. |
| 26 | pub const REPO_CHANGED: &str = "repo_changed"; |
| 27 | |
| 28 | /// What a run's job starting or finishing is told as: not an event, as |
| 29 | /// nothing else hears of it. |
| 30 | pub const RUN_UPDATED: &str = "run.updated"; |
| 31 | |
| 32 | /// The most changes one call carries; a busier batch sends the rest in more. |
| 33 | pub const MAX_CHANGES: usize = 200; |
| 34 | |
| 35 | /// Something that happened in a repository, as live pages hear of it. |
| 36 | #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] |
| 37 | pub struct RepoChange { |
| 38 | pub repo_id: String, |
| 39 | /// The event's type (`git.push`, `pull.updated`, `checks.completed`, …), |
| 40 | /// or [`RUN_UPDATED`]. |
| 41 | pub kind: String, |
| 42 | /// The full ref a push or deletion moved (`refs/heads/main`), or a run's. |
| 43 | #[serde(rename = "ref", default, skip_serializing_if = "Option::is_none")] |
| 44 | pub git_ref: Option<String>, |
| 45 | /// The work item or pull request it is on, or the pull request a run ran for. |
| 46 | #[serde(default, skip_serializing_if = "Option::is_none")] |
| 47 | pub number: Option<u32>, |
| 48 | /// The commit: where a push left its ref, the one checked or reported on. |
| 49 | #[serde(default, skip_serializing_if = "Option::is_none")] |
| 50 | pub sha: Option<String>, |
| 51 | /// A workflow run's id. |
| 52 | #[serde(default, skip_serializing_if = "Option::is_none")] |
| 53 | pub run: Option<String>, |
| 54 | /// How many commits a push brought to its branch, when it was counted. |
| 55 | #[serde(default, skip_serializing_if = "Option::is_none")] |
| 56 | pub commits: Option<u32>, |
| 57 | /// RFC 3339. |
| 58 | pub at: String, |
| 59 | } |
| 60 | |
| 61 | /// `repo_changed`. |
| 62 | #[derive(Clone, Debug, Default, Serialize, Deserialize)] |
| 63 | pub struct RepoChangedArgs { |
| 64 | pub changes: Vec<RepoChange>, |
| 65 | } |
| 66 | |
| 67 | /// The events a repository page shows: exact types, and families ending |
| 68 | /// in a dot. Anything else (reads, fetches, settings, audit) changes no |
| 69 | /// page anyone is looking at. |
| 70 | const LIVE: [&str; 28] = [ |
| 71 | "git.push", |
| 72 | "git.deleted", |
| 73 | "branch.renamed", |
| 74 | "repo.default_branch_changed", |
| 75 | "pull.", |
| 76 | "work_item.", |
| 77 | "comment.", |
| 78 | "review.", |
| 79 | "checks.", |
| 80 | "check_run.", |
| 81 | "check_suite.", |
| 82 | "status.created", |
| 83 | "workflow.", |
| 84 | "release.", |
| 85 | "queue.changed", |
| 86 | "deployment.", |
| 87 | "deployment_status.", |
| 88 | "secret_scanning_alert.", |
| 89 | "code_scanning_alert.", |
| 90 | "vulnerability_alert.", |
| 91 | // Coordinating work: the work item, pull request and Work pages show them. |
| 92 | "work.", |
| 93 | "session.started", |
| 94 | "session.ended", |
| 95 | // A session recorded more: at most every few seconds each, for the |
| 96 | // pages showing it (not every entry, which `session.appended` is). |
| 97 | "session.progressed", |
| 98 | "dependency.changed", |
| 99 | "overlap.detected", |
| 100 | "integration.", |
| 101 | "decision.recorded", |
| 102 | ]; |
| 103 | |
| 104 | /// Whether an event of type `kind` changes a repository page. |
| 105 | pub fn is_live(kind: &str) -> bool { |
| 106 | LIVE.iter().any(|live| if live.ends_with('.') { kind.starts_with(live) } else { kind == *live }) |
| 107 | } |
| 108 | |
| 109 | fn text(value: &serde_json::Value) -> Option<String> { |
| 110 | value.as_str().filter(|text| !text.is_empty()).map(|text| text.chars().take(200).collect()) |
| 111 | } |
| 112 | |
| 113 | fn number(value: &serde_json::Value) -> Option<u32> { |
| 114 | value.as_u64().and_then(|n| u32::try_from(n).ok()) |
| 115 | } |
| 116 | |
| 117 | /// What live pages hear of an event, or None when it is not about a |
| 118 | /// repository or changes no page. |
| 119 | pub fn change_of(event: &Event) -> Option<RepoChange> { |
| 120 | if !is_live(&event.kind) { |
| 121 | return None; |
| 122 | } |
| 123 | let data = &event.data; |
| 124 | let repo_id = event.repo_id.clone().filter(|id| !id.is_empty()).or_else(|| text(&data["repoId"]))?; |
| 125 | let family = event.kind.split('.').next().unwrap_or_default(); |
| 126 | let mut change = RepoChange { repo_id, kind: event.kind.clone(), at: event.time.clone(), ..RepoChange::default() }; |
| 127 | match family { |
| 128 | "git" => { |
| 129 | change.git_ref = text(&data["ref"]); |
| 130 | change.sha = text(&data["after"]); |
| 131 | change.commits = number(&data["commits"]); |
| 132 | } |
| 133 | "workflow" => { |
| 134 | change.git_ref = text(&data["ref"]); |
| 135 | change.sha = text(&data["sha"]); |
| 136 | change.run = text(&data["runId"]); |
| 137 | change.number = number(&data["pull"]); |
| 138 | } |
| 139 | "checks" => { |
| 140 | change.number = number(&data["number"]); |
| 141 | change.sha = text(&data["commit"]); |
| 142 | } |
| 143 | "status" => change.sha = text(&data["sha"]), |
| 144 | "check_run" => change.sha = text(&data["checkRun"]["headSha"]), |
| 145 | "check_suite" => change.sha = text(&data["checkSuite"]["headSha"]), |
| 146 | "code_scanning_alert" => change.number = number(&data["alertNumber"]), |
| 147 | "secret_scanning_alert" | "vulnerability_alert" | "release" | "queue" | "deployment" | "deployment_status" | "repo" | "branch" => {} |
| 148 | // Work items, pull requests, comments and reviews name their number. |
| 149 | _ => change.number = number(&data["number"]), |
| 150 | } |
| 151 | Some(change) |
| 152 | } |
| 153 | |
| 154 | /// The notify service's method that takes [`WorkspaceChangedArgs`]. |
| 155 | pub const WORKSPACE_CHANGED: &str = "workspace_changed"; |
| 156 | |
| 157 | /// Something that happened to a workspace's work items, as its Work pages |
| 158 | /// hear of it: what kind of thing moved, and the work item's number. |
| 159 | #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] |
| 160 | pub struct WorkspaceChange { |
| 161 | /// Its slug, lowercase. |
| 162 | pub workspace: String, |
| 163 | /// The event's type (`work_item.updated`, `comment.created`, `work.claimed`, …). |
| 164 | pub kind: String, |
| 165 | /// The work item it is on. |
| 166 | #[serde(default, skip_serializing_if = "Option::is_none")] |
| 167 | pub number: Option<u32>, |
| 168 | /// RFC 3339. |
| 169 | pub at: String, |
| 170 | } |
| 171 | |
| 172 | /// `workspace_changed`. |
| 173 | #[derive(Clone, Debug, Default, Serialize, Deserialize)] |
| 174 | pub struct WorkspaceChangedArgs { |
| 175 | pub changes: Vec<WorkspaceChange>, |
| 176 | } |
| 177 | |
| 178 | /// What a workspace's Work pages hear of an event, or None when it is not |
| 179 | /// about one of its work items: a work item's own events, and the events |
| 180 | /// that name one (its comments and reactions, claims on it, what it is |
| 181 | /// blocked by). |
| 182 | pub fn workspace_change_of(event: &Event) -> Option<WorkspaceChange> { |
| 183 | if !is_live(&event.kind) { |
| 184 | return None; |
| 185 | } |
| 186 | let data = &event.data; |
| 187 | let workspace = text(&data["workspace"])?.to_lowercase(); |
| 188 | let about_item = event.kind.starts_with("work_item.") |
| 189 | || text(&data["itemId"]).is_some() |
| 190 | || matches!(event.kind.as_str(), "dependency.changed" | "work.claimed" | "work.released" | "work.handoff_requested"); |
| 191 | if !about_item { |
| 192 | return None; |
| 193 | } |
| 194 | Some(WorkspaceChange { |
| 195 | workspace, |
| 196 | kind: event.kind.clone(), |
| 197 | number: number(&data["number"]).or_else(|| number(&data["item"])), |
| 198 | at: event.time.clone(), |
| 199 | }) |
| 200 | } |
| 201 | |
| 202 | /// The workspace changes a batch of events makes, in order, in calls of at |
| 203 | /// most [`MAX_CHANGES`]. |
| 204 | pub fn workspace_changes_of(events: &[Event]) -> Vec<Vec<WorkspaceChange>> { |
| 205 | let changes: Vec<WorkspaceChange> = events.iter().filter_map(workspace_change_of).collect(); |
| 206 | changes.chunks(MAX_CHANGES).map(<[WorkspaceChange]>::to_vec).collect() |
| 207 | } |
| 208 | |
| 209 | /// The changes a batch of events makes, in order, in calls of at most |
| 210 | /// [`MAX_CHANGES`]. |
| 211 | pub fn changes_of(events: &[Event]) -> Vec<Vec<RepoChange>> { |
| 212 | let changes: Vec<RepoChange> = events.iter().filter_map(change_of).collect(); |
| 213 | changes.chunks(MAX_CHANGES).map(<[RepoChange]>::to_vec).collect() |
| 214 | } |
| 215 | |
| 216 | #[cfg(test)] |
| 217 | mod tests { |
| 218 | use super::*; |
| 219 | use serde_json::json; |
| 220 | |
| 221 | fn event(kind: &str, repo_id: Option<&str>, data: serde_json::Value) -> Event { |
| 222 | Event { |
| 223 | id: "evt_1".into(), |
| 224 | kind: kind.into(), |
| 225 | source: "test".into(), |
| 226 | time: "2026-10-11T10:00:00Z".into(), |
| 227 | repo_id: repo_id.map(Into::into), |
| 228 | actor: Some("usr_1".into()), |
| 229 | data, |
| 230 | } |
| 231 | } |
| 232 | |
| 233 | #[test] |
| 234 | fn a_push_says_its_ref_where_it_went_and_how_many_commits() { |
| 235 | let push = event( |
| 236 | "git.push", |
| 237 | Some("rep_1"), |
| 238 | json!({ "repoId": "rep_1", "ref": "refs/heads/main", "before": "a", "after": "b", "defaultBranch": true, "commits": 3 }), |
| 239 | ); |
| 240 | assert_eq!( |
| 241 | change_of(&push), |
| 242 | Some(RepoChange { |
| 243 | repo_id: "rep_1".into(), |
| 244 | kind: "git.push".into(), |
| 245 | git_ref: Some("refs/heads/main".into()), |
| 246 | sha: Some("b".into()), |
| 247 | commits: Some(3), |
| 248 | at: "2026-10-11T10:00:00Z".into(), |
| 249 | ..RepoChange::default() |
| 250 | }) |
| 251 | ); |
| 252 | } |
| 253 | |
| 254 | #[test] |
| 255 | fn a_workspaces_work_pages_hear_of_its_work_items_only() { |
| 256 | let data = json!({ "itemId": "wki_1", "workspace": "Acme", "number": 40, "repoIds": [], "title": "secret plans" }); |
| 257 | let change = workspace_change_of(&event("work_item.updated", None, data)).unwrap(); |
| 258 | assert_eq!( |
| 259 | change, |
| 260 | WorkspaceChange { workspace: "acme".into(), kind: "work_item.updated".into(), number: Some(40), at: "2026-10-11T10:00:00Z".into() } |
| 261 | ); |
| 262 | // A comment on it, and a claim (which names it as `item`). |
| 263 | let comment = json!({ "itemId": "wki_1", "workspace": "acme", "number": 40, "commentId": "cmt_1" }); |
| 264 | assert_eq!(workspace_change_of(&event("comment.created", None, comment)).unwrap().number, Some(40)); |
| 265 | let claim = json!({ "workspace": "acme", "number": 40, "item": 40, "claim": {} }); |
| 266 | assert!(workspace_change_of(&event("work.claimed", None, claim)).is_some()); |
| 267 | // A pull request's comment, and what names no workspace, are not. |
| 268 | let on_pull = json!({ "repoId": "rep_1", "number": 4, "pullId": "pr_4" }); |
| 269 | assert!(workspace_change_of(&event("comment.created", Some("rep_1"), on_pull)).is_none()); |
| 270 | assert!(workspace_change_of(&event("work_item.updated", None, json!({ "number": 4 }))).is_none()); |
| 271 | assert!(workspace_change_of(&event("audit.recorded", None, json!({ "workspace": "acme", "itemId": "wki_1" }))).is_none()); |
| 272 | } |
| 273 | |
| 274 | #[test] |
| 275 | fn pull_requests_work_items_comments_and_checks_name_their_number() { |
| 276 | for kind in ["pull.updated", "pull.merged", "work_item.closed", "comment.created", "review.completed"] { |
| 277 | let change = change_of(&event(kind, Some("rep_1"), json!({ "number": 7, "title": "secret plans" }))).unwrap(); |
| 278 | assert_eq!(change.number, Some(7), "{kind}"); |
| 279 | } |
| 280 | let checks = change_of(&event("checks.completed", Some("rep_1"), json!({ "number": 4, "commit": "abc", "status": "passed" }))).unwrap(); |
| 281 | assert_eq!((checks.number, checks.sha.as_deref()), (Some(4), Some("abc"))); |
| 282 | } |
| 283 | |
| 284 | #[test] |
| 285 | fn runs_name_their_id_commit_and_pull_request() { |
| 286 | let run = change_of(&event( |
| 287 | "workflow.completed", |
| 288 | Some("rep_1"), |
| 289 | json!({ "runId": "run_9", "number": 12, "ref": "refs/heads/x", "sha": "def", "pull": 5, "conclusion": "success" }), |
| 290 | )) |
| 291 | .unwrap(); |
| 292 | assert_eq!(run.run.as_deref(), Some("run_9")); |
| 293 | assert_eq!(run.number, Some(5), "the pull request, not the run's own number"); |
| 294 | assert_eq!(run.sha.as_deref(), Some("def")); |
| 295 | let status = change_of(&event("status.created", Some("rep_1"), json!({ "sha": "def", "context": "ci", "state": "pending" }))).unwrap(); |
| 296 | assert_eq!(status.sha.as_deref(), Some("def")); |
| 297 | let check = change_of(&event("check_run.completed", Some("rep_1"), json!({ "checkRun": { "headSha": "fff" } }))).unwrap(); |
| 298 | assert_eq!(check.sha.as_deref(), Some("fff")); |
| 299 | } |
| 300 | |
| 301 | #[test] |
| 302 | fn nothing_written_in_the_repository_goes_along() { |
| 303 | let change = change_of(&event("work_item.opened", Some("rep_1"), json!({ "number": 1, "title": "Leaked title", "body": "Leaked body" }))).unwrap(); |
| 304 | let sent = serde_json::to_string(&change).unwrap(); |
| 305 | assert!(!sent.contains("Leaked")); |
| 306 | assert_eq!(sent, r#"{"repo_id":"rep_1","kind":"work_item.opened","number":1,"at":"2026-10-11T10:00:00Z"}"#); |
| 307 | } |
| 308 | |
| 309 | #[test] |
| 310 | fn only_what_a_page_shows_and_only_with_a_repository() { |
| 311 | assert!(change_of(&event("git.fetch", Some("rep_1"), json!({}))).is_none()); |
| 312 | assert!(change_of(&event("repo.collaborator_added", Some("rep_1"), json!({}))).is_none()); |
| 313 | assert!(change_of(&event("pull.opened", None, json!({ "number": 1 }))).is_none()); |
| 314 | // The repository from the data when the event has none. |
| 315 | assert_eq!(change_of(&event("pull.opened", None, json!({ "repoId": "rep_2", "number": 1 }))).unwrap().repo_id, "rep_2"); |
| 316 | for kind in ["git.push", "git.deleted", "release.published", "secret_scanning_alert.created", "queue.changed", "deployment_status.created"] { |
| 317 | assert!(is_live(kind), "{kind}"); |
| 318 | } |
| 319 | for kind in ["git.pushed", "repo.created", "user.deleted", "workspace.renamed"] { |
| 320 | assert!(!is_live(kind), "{kind}"); |
| 321 | } |
| 322 | } |
| 323 | |
| 324 | #[test] |
| 325 | fn coordination_names_the_work_item_or_pull_request_it_is_about() { |
| 326 | for kind in ["work.claimed", "work.claim_expired", "session.started", "dependency.changed", "overlap.detected", "integration.failed", "integration.passed", "decision.recorded"] { |
| 327 | let change = change_of(&event(kind, None, json!({ "repoId": "rep_1", "number": 12, "item": 4 }))).unwrap(); |
| 328 | assert_eq!((change.repo_id.as_str(), change.number), ("rep_1", Some(12)), "{kind}"); |
| 329 | } |
| 330 | // Each recorded step of a session changes no page anyone is watching live. |
| 331 | assert!(change_of(&event("session.appended", Some("rep_1"), json!({ "number": 12 }))).is_none()); |
| 332 | // Its progress, at most every few seconds, does: the session's page follows it. |
| 333 | let progressed = change_of(&event("session.progressed", None, json!({ "repoId": "rep_1", "number": 12, "sessionId": "ses_1" }))).unwrap(); |
| 334 | assert_eq!((progressed.kind.as_str(), progressed.number), ("session.progressed", Some(12))); |
| 335 | } |
| 336 | |
| 337 | #[test] |
| 338 | fn a_busy_batch_goes_in_several_calls() { |
| 339 | let events: Vec<Event> = (0..450).map(|n| event("pull.updated", Some("rep_1"), json!({ "number": n }))).collect(); |
| 340 | let calls = changes_of(&events); |
| 341 | assert_eq!(calls.iter().map(Vec::len).collect::<Vec<_>>(), vec![200, 200, 50]); |
| 342 | assert!(changes_of(&[event("git.fetch", Some("rep_1"), json!({}))]).is_empty()); |
| 343 | } |
| 344 | } |