| 1 | //! Live repository pages: every event a repository page shows is passed to |
| 2 | //! the notify service as a small change (`g1t_contracts::live`), which |
| 3 | //! tells the people watching that repository's pages. One call per batch |
| 4 | //! (more for a very busy one); nothing written in the repository goes |
| 5 | //! along, and who may see it is notify's to keep: a tab watches a |
| 6 | //! 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. |
| 10 | |
| 11 | use g1t_contracts::events::Event; |
| 12 | use g1t_contracts::live::{MAX_CHANGES, REPO_CHANGED, RepoChange, RepoChangedArgs, change_of}; |
| 13 | use worker::{Fetcher, Result}; |
| 14 | |
| 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 | |
| 35 | /// Tells notify of the batch's changes. Logged and left when it cannot be |
| 36 | /// reached: pages then catch up by themselves, and nothing else waits on it. |
| 37 | pub(crate) async fn tell(notify: &Fetcher, events: &[Event]) { |
| 38 | for changes in changes_of(events) { |
| 39 | let told: Result<serde_json::Value> = g1t_kit::call(notify, REPO_CHANGED, &RepoChangedArgs { changes }).await; |
| 40 | if let Err(error) = told { |
| 41 | worker::console_error!("events: live pages not told: {error}"); |
| 42 | } |
| 43 | } |
| 44 | } |
| 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 | } |