| 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 | use serde::{Deserialize, Serialize}; |
| 16 | |
| 17 | use crate::events::Event; |
| 18 | |
| 19 | /// The notify service's method that takes [`RepoChangedArgs`]. |
| 20 | pub const REPO_CHANGED: &str = "repo_changed"; |
| 21 | |
| 22 | /// What a run's job starting or finishing is told as: not an event, as |
| 23 | /// nothing else hears of it. |
| 24 | pub const RUN_UPDATED: &str = "run.updated"; |
| 25 | |
| 26 | /// The most changes one call carries; a busier batch sends the rest in more. |
| 27 | pub const MAX_CHANGES: usize = 200; |
| 28 | |
| 29 | /// Something that happened in a repository, as live pages hear of it. |
| 30 | #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] |
| 31 | pub struct RepoChange { |
| 32 | pub repo_id: String, |
| 33 | /// The event's type (`git.push`, `pull.updated`, `checks.completed`, …), |
| 34 | /// or [`RUN_UPDATED`]. |
| 35 | pub kind: String, |
| 36 | /// The full ref a push or deletion moved (`refs/heads/main`), or a run's. |
| 37 | #[serde(rename = "ref", default, skip_serializing_if = "Option::is_none")] |
| 38 | pub git_ref: Option<String>, |
| 39 | /// The issue or pull request it is on, or the pull request a run ran for. |
| 40 | #[serde(default, skip_serializing_if = "Option::is_none")] |
| 41 | pub number: Option<u32>, |
| 42 | /// The commit: where a push left its ref, the one checked or reported on. |
| 43 | #[serde(default, skip_serializing_if = "Option::is_none")] |
| 44 | pub sha: Option<String>, |
| 45 | /// A workflow run's id. |
| 46 | #[serde(default, skip_serializing_if = "Option::is_none")] |
| 47 | pub run: Option<String>, |
| 48 | /// How many commits a push brought to its branch, when it was counted. |
| 49 | #[serde(default, skip_serializing_if = "Option::is_none")] |
| 50 | pub commits: Option<u32>, |
| 51 | /// RFC 3339. |
| 52 | pub at: String, |
| 53 | } |
| 54 | |
| 55 | /// `repo_changed`. |
| 56 | #[derive(Clone, Debug, Default, Serialize, Deserialize)] |
| 57 | pub struct RepoChangedArgs { |
| 58 | pub changes: Vec<RepoChange>, |
| 59 | } |
| 60 | |
| 61 | /// The events a repository page shows: exact types, and families ending |
| 62 | /// in a dot. Anything else (reads, fetches, settings, audit) changes no |
| 63 | /// page anyone is looking at. |
| 64 | const LIVE: [&str; 28] = [ |
| 65 | "git.push", |
| 66 | "git.deleted", |
| 67 | "branch.renamed", |
| 68 | "repo.default_branch_changed", |
| 69 | "pull.", |
| 70 | "issue.", |
| 71 | "comment.", |
| 72 | "review.", |
| 73 | "checks.", |
| 74 | "check_run.", |
| 75 | "check_suite.", |
| 76 | "status.created", |
| 77 | "workflow.", |
| 78 | "release.", |
| 79 | "queue.changed", |
| 80 | "deployment.", |
| 81 | "deployment_status.", |
| 82 | "secret_scanning_alert.", |
| 83 | "code_scanning_alert.", |
| 84 | "vulnerability_alert.", |
| 85 | // Coordinating work: the issue, pull request and Work pages show them. |
| 86 | "work.", |
| 87 | "session.started", |
| 88 | "session.ended", |
| 89 | // A session recorded more: at most every few seconds each, for the |
| 90 | // pages showing it (not every entry, which `session.appended` is). |
| 91 | "session.progressed", |
| 92 | "dependency.changed", |
| 93 | "overlap.detected", |
| 94 | "integration.", |
| 95 | "decision.recorded", |
| 96 | ]; |
| 97 | |
| 98 | /// Whether an event of type `kind` changes a repository page. |
| 99 | pub fn is_live(kind: &str) -> bool { |
| 100 | LIVE.iter().any(|live| if live.ends_with('.') { kind.starts_with(live) } else { kind == *live }) |
| 101 | } |
| 102 | |
| 103 | fn text(value: &serde_json::Value) -> Option<String> { |
| 104 | value.as_str().filter(|text| !text.is_empty()).map(|text| text.chars().take(200).collect()) |
| 105 | } |
| 106 | |
| 107 | fn number(value: &serde_json::Value) -> Option<u32> { |
| 108 | value.as_u64().and_then(|n| u32::try_from(n).ok()) |
| 109 | } |
| 110 | |
| 111 | /// What live pages hear of an event, or None when it is not about a |
| 112 | /// repository or changes no page. |
| 113 | pub fn change_of(event: &Event) -> Option<RepoChange> { |
| 114 | if !is_live(&event.kind) { |
| 115 | return None; |
| 116 | } |
| 117 | let data = &event.data; |
| 118 | let repo_id = event.repo_id.clone().filter(|id| !id.is_empty()).or_else(|| text(&data["repoId"]))?; |
| 119 | let family = event.kind.split('.').next().unwrap_or_default(); |
| 120 | let mut change = RepoChange { repo_id, kind: event.kind.clone(), at: event.time.clone(), ..RepoChange::default() }; |
| 121 | match family { |
| 122 | "git" => { |
| 123 | change.git_ref = text(&data["ref"]); |
| 124 | change.sha = text(&data["after"]); |
| 125 | change.commits = number(&data["commits"]); |
| 126 | } |
| 127 | "workflow" => { |
| 128 | change.git_ref = text(&data["ref"]); |
| 129 | change.sha = text(&data["sha"]); |
| 130 | change.run = text(&data["runId"]); |
| 131 | change.number = number(&data["pull"]); |
| 132 | } |
| 133 | "checks" => { |
| 134 | change.number = number(&data["number"]); |
| 135 | change.sha = text(&data["commit"]); |
| 136 | } |
| 137 | "status" => change.sha = text(&data["sha"]), |
| 138 | "check_run" => change.sha = text(&data["checkRun"]["headSha"]), |
| 139 | "check_suite" => change.sha = text(&data["checkSuite"]["headSha"]), |
| 140 | "code_scanning_alert" => change.number = number(&data["alertNumber"]), |
| 141 | "secret_scanning_alert" | "vulnerability_alert" | "release" | "queue" | "deployment" | "deployment_status" | "repo" | "branch" => {} |
| 142 | // Issues, pull requests, comments and reviews name their number. |
| 143 | _ => change.number = number(&data["number"]), |
| 144 | } |
| 145 | Some(change) |
| 146 | } |
| 147 | |
| 148 | /// The changes a batch of events makes, in order, in calls of at most |
| 149 | /// [`MAX_CHANGES`]. |
| 150 | pub fn changes_of(events: &[Event]) -> Vec<Vec<RepoChange>> { |
| 151 | let changes: Vec<RepoChange> = events.iter().filter_map(change_of).collect(); |
| 152 | changes.chunks(MAX_CHANGES).map(<[RepoChange]>::to_vec).collect() |
| 153 | } |
| 154 | |
| 155 | #[cfg(test)] |
| 156 | mod tests { |
| 157 | use super::*; |
| 158 | use serde_json::json; |
| 159 | |
| 160 | fn event(kind: &str, repo_id: Option<&str>, data: serde_json::Value) -> Event { |
| 161 | Event { |
| 162 | id: "evt_1".into(), |
| 163 | kind: kind.into(), |
| 164 | source: "test".into(), |
| 165 | time: "2026-10-11T10:00:00Z".into(), |
| 166 | repo_id: repo_id.map(Into::into), |
| 167 | actor: Some("usr_1".into()), |
| 168 | data, |
| 169 | } |
| 170 | } |
| 171 | |
| 172 | #[test] |
| 173 | fn a_push_says_its_ref_where_it_went_and_how_many_commits() { |
| 174 | let push = event( |
| 175 | "git.push", |
| 176 | Some("rep_1"), |
| 177 | json!({ "repoId": "rep_1", "ref": "refs/heads/main", "before": "a", "after": "b", "defaultBranch": true, "commits": 3 }), |
| 178 | ); |
| 179 | assert_eq!( |
| 180 | change_of(&push), |
| 181 | Some(RepoChange { |
| 182 | repo_id: "rep_1".into(), |
| 183 | kind: "git.push".into(), |
| 184 | git_ref: Some("refs/heads/main".into()), |
| 185 | sha: Some("b".into()), |
| 186 | commits: Some(3), |
| 187 | at: "2026-10-11T10:00:00Z".into(), |
| 188 | ..RepoChange::default() |
| 189 | }) |
| 190 | ); |
| 191 | } |
| 192 | |
| 193 | #[test] |
| 194 | fn pull_requests_issues_comments_and_checks_name_their_number() { |
| 195 | for kind in ["pull.updated", "pull.merged", "issue.closed", "comment.created", "review.completed"] { |
| 196 | let change = change_of(&event(kind, Some("rep_1"), json!({ "number": 7, "title": "secret plans" }))).unwrap(); |
| 197 | assert_eq!(change.number, Some(7), "{kind}"); |
| 198 | } |
| 199 | let checks = change_of(&event("checks.completed", Some("rep_1"), json!({ "number": 4, "commit": "abc", "status": "passed" }))).unwrap(); |
| 200 | assert_eq!((checks.number, checks.sha.as_deref()), (Some(4), Some("abc"))); |
| 201 | } |
| 202 | |
| 203 | #[test] |
| 204 | fn runs_name_their_id_commit_and_pull_request() { |
| 205 | let run = change_of(&event( |
| 206 | "workflow.completed", |
| 207 | Some("rep_1"), |
| 208 | json!({ "runId": "run_9", "number": 12, "ref": "refs/heads/x", "sha": "def", "pull": 5, "conclusion": "success" }), |
| 209 | )) |
| 210 | .unwrap(); |
| 211 | assert_eq!(run.run.as_deref(), Some("run_9")); |
| 212 | assert_eq!(run.number, Some(5), "the pull request, not the run's own number"); |
| 213 | assert_eq!(run.sha.as_deref(), Some("def")); |
| 214 | let status = change_of(&event("status.created", Some("rep_1"), json!({ "sha": "def", "context": "ci", "state": "pending" }))).unwrap(); |
| 215 | assert_eq!(status.sha.as_deref(), Some("def")); |
| 216 | let check = change_of(&event("check_run.completed", Some("rep_1"), json!({ "checkRun": { "headSha": "fff" } }))).unwrap(); |
| 217 | assert_eq!(check.sha.as_deref(), Some("fff")); |
| 218 | } |
| 219 | |
| 220 | #[test] |
| 221 | fn nothing_written_in_the_repository_goes_along() { |
| 222 | let change = change_of(&event("issue.opened", Some("rep_1"), json!({ "number": 1, "title": "Leaked title", "body": "Leaked body" }))).unwrap(); |
| 223 | let sent = serde_json::to_string(&change).unwrap(); |
| 224 | assert!(!sent.contains("Leaked")); |
| 225 | assert_eq!(sent, r#"{"repo_id":"rep_1","kind":"issue.opened","number":1,"at":"2026-10-11T10:00:00Z"}"#); |
| 226 | } |
| 227 | |
| 228 | #[test] |
| 229 | fn only_what_a_page_shows_and_only_with_a_repository() { |
| 230 | assert!(change_of(&event("git.fetch", Some("rep_1"), json!({}))).is_none()); |
| 231 | assert!(change_of(&event("repo.collaborator_added", Some("rep_1"), json!({}))).is_none()); |
| 232 | assert!(change_of(&event("pull.opened", None, json!({ "number": 1 }))).is_none()); |
| 233 | // The repository from the data when the event has none. |
| 234 | assert_eq!(change_of(&event("pull.opened", None, json!({ "repoId": "rep_2", "number": 1 }))).unwrap().repo_id, "rep_2"); |
| 235 | for kind in ["git.push", "git.deleted", "release.published", "secret_scanning_alert.created", "queue.changed", "deployment_status.created"] { |
| 236 | assert!(is_live(kind), "{kind}"); |
| 237 | } |
| 238 | for kind in ["git.pushed", "repo.created", "user.deleted", "workspace.renamed"] { |
| 239 | assert!(!is_live(kind), "{kind}"); |
| 240 | } |
| 241 | } |
| 242 | |
| 243 | #[test] |
| 244 | fn coordination_names_the_issue_or_pull_request_it_is_about() { |
| 245 | for kind in ["work.claimed", "work.claim_expired", "session.started", "dependency.changed", "overlap.detected", "integration.failed", "integration.passed", "decision.recorded"] { |
| 246 | let change = change_of(&event(kind, None, json!({ "repoId": "rep_1", "number": 12, "issue": 4 }))).unwrap(); |
| 247 | assert_eq!((change.repo_id.as_str(), change.number), ("rep_1", Some(12)), "{kind}"); |
| 248 | } |
| 249 | // Each recorded step of a session changes no page anyone is watching live. |
| 250 | assert!(change_of(&event("session.appended", Some("rep_1"), json!({ "number": 12 }))).is_none()); |
| 251 | // Its progress, at most every few seconds, does: the session's page follows it. |
| 252 | let progressed = change_of(&event("session.progressed", None, json!({ "repoId": "rep_1", "number": 12, "sessionId": "ses_1" }))).unwrap(); |
| 253 | assert_eq!((progressed.kind.as_str(), progressed.number), ("session.progressed", Some(12))); |
| 254 | } |
| 255 | |
| 256 | #[test] |
| 257 | fn a_busy_batch_goes_in_several_calls() { |
| 258 | let events: Vec<Event> = (0..450).map(|n| event("pull.updated", Some("rep_1"), json!({ "number": n }))).collect(); |
| 259 | let calls = changes_of(&events); |
| 260 | assert_eq!(calls.iter().map(Vec::len).collect::<Vec<_>>(), vec![200, 200, 50]); |
| 261 | assert!(changes_of(&[event("git.fetch", Some("rep_1"), json!({}))]).is_empty()); |
| 262 | } |
| 263 | } |