| 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 its origin (the event's own |
| 9 | //! repository, which it may not have) and to its workspace's Work pages |
| 10 | //! (`workspace_changed`), which hear of it with or without a repository. |
| 11 | //! The other repositories it is linked to (`repoIds`) hear of it only when |
| 12 | //! their readers may all read it: when its origin is public. A private |
| 13 | //! one's kind and number are not for whoever reads a repository it only |
| 14 | //! names, nor is an item with no origin, which is its workspace's alone. |
| 15 | |
| 16 | use g1t_contracts::events::Event; |
| 17 | use g1t_contracts::live::{ |
| 18 | MAX_CHANGES, REPO_CHANGED, RepoChange, RepoChangedArgs, WORKSPACE_CHANGED, WorkspaceChangedArgs, change_of, workspace_changes_of, |
| 19 | }; |
| 20 | use g1t_contracts::repos::{MAX_READABLE, ReadableArgs, Repo}; |
| 21 | use std::collections::HashSet; |
| 22 | use worker::{Fetcher, Result}; |
| 23 | |
| 24 | /// The changes an event makes: one for its repository, and one for each |
| 25 | /// other repository it names in `repoIds` when the event's repository (a |
| 26 | /// work item's origin) is among the `public` ones. |
| 27 | pub(crate) fn changes_for(event: &Event, public: &HashSet<String>) -> Vec<RepoChange> { |
| 28 | let linked_to = event.data["repoIds"].as_array().is_some_and(|ids| !ids.is_empty()); |
| 29 | let origin = event.repo_id.as_deref().filter(|id| !id.is_empty()); |
| 30 | // A work item's origin is the event's own repository, never a |
| 31 | // `repoId` in its data. |
| 32 | let mut changes: Vec<RepoChange> = if linked_to && origin.is_none() { |
| 33 | Vec::new() |
| 34 | } else { |
| 35 | change_of(event).into_iter().collect() |
| 36 | }; |
| 37 | if !origin.is_some_and(|origin| public.contains(origin)) { |
| 38 | return changes; |
| 39 | } |
| 40 | for repo_id in event.data["repoIds"].as_array().into_iter().flatten().filter_map(|id| id.as_str()) { |
| 41 | if repo_id.is_empty() || changes.iter().any(|change| change.repo_id == repo_id) { |
| 42 | continue; |
| 43 | } |
| 44 | let linked = Event { repo_id: Some(repo_id.to_owned()), ..event.clone() }; |
| 45 | changes.extend(change_of(&linked)); |
| 46 | } |
| 47 | changes |
| 48 | } |
| 49 | |
| 50 | /// A batch's changes, in order, in calls of at most [`MAX_CHANGES`]. |
| 51 | pub(crate) fn changes_of(events: &[Event], public: &HashSet<String>) -> Vec<Vec<RepoChange>> { |
| 52 | let changes: Vec<RepoChange> = events.iter().flat_map(|event| changes_for(event, public)).collect(); |
| 53 | changes.chunks(MAX_CHANGES).map(<[RepoChange]>::to_vec).collect() |
| 54 | } |
| 55 | |
| 56 | /// Which of these repositories anyone may read, in one `readable` call |
| 57 | /// with no viewer. None when the repos service cannot say: nothing is |
| 58 | /// taken to be public that is not known to be. |
| 59 | pub(crate) async fn public_of(repos: &Fetcher, ids: Vec<String>) -> HashSet<String> { |
| 60 | if ids.is_empty() { |
| 61 | return HashSet::new(); |
| 62 | } |
| 63 | let readable: Result<Vec<Repo>> = g1t_kit::call(repos, "readable", &ReadableArgs { ids, viewer: None }).await; |
| 64 | match readable { |
| 65 | Ok(repos) => repos.into_iter().map(|repo| repo.id).collect(), |
| 66 | Err(error) => { |
| 67 | worker::console_error!("events: could not tell which repositories are public: {error}"); |
| 68 | HashSet::new() |
| 69 | } |
| 70 | } |
| 71 | } |
| 72 | |
| 73 | /// The origins of the batch's events that name other repositories |
| 74 | /// (`repoIds`): the ones whose publicness decides who else hears. |
| 75 | fn origins_linked(events: &[Event]) -> Vec<String> { |
| 76 | let origins: HashSet<String> = events |
| 77 | .iter() |
| 78 | .filter(|event| event.data["repoIds"].as_array().is_some_and(|ids| ids.len() > 1)) |
| 79 | .filter_map(|event| event.repo_id.clone().filter(|id| !id.is_empty())) |
| 80 | .collect(); |
| 81 | origins.into_iter().take(MAX_READABLE).collect() |
| 82 | } |
| 83 | |
| 84 | /// Tells notify of the batch's changes. Logged and left when it cannot be |
| 85 | /// reached: pages then catch up by themselves, and nothing else waits on it. |
| 86 | pub(crate) async fn tell(notify: &Fetcher, repos: &Fetcher, events: &[Event]) { |
| 87 | let public = public_of(repos, origins_linked(events)).await; |
| 88 | for changes in changes_of(events, &public) { |
| 89 | let told: Result<serde_json::Value> = g1t_kit::call(notify, REPO_CHANGED, &RepoChangedArgs { changes }).await; |
| 90 | if let Err(error) = told { |
| 91 | worker::console_error!("events: live pages not told: {error}"); |
| 92 | } |
| 93 | } |
| 94 | for changes in workspace_changes_of(events) { |
| 95 | let told: Result<serde_json::Value> = g1t_kit::call(notify, WORKSPACE_CHANGED, &WorkspaceChangedArgs { changes }).await; |
| 96 | if let Err(error) = told { |
| 97 | worker::console_error!("events: live Work pages not told: {error}"); |
| 98 | } |
| 99 | } |
| 100 | } |
| 101 | |
| 102 | #[cfg(test)] |
| 103 | mod tests { |
| 104 | use super::*; |
| 105 | use serde_json::json; |
| 106 | |
| 107 | fn event(kind: &str, repo_id: Option<&str>, data: serde_json::Value) -> Event { |
| 108 | Event { |
| 109 | id: "evt_1".into(), |
| 110 | kind: kind.into(), |
| 111 | source: "work".into(), |
| 112 | time: "2026-10-11T10:00:00Z".into(), |
| 113 | repo_id: repo_id.map(Into::into), |
| 114 | actor: None, |
| 115 | data, |
| 116 | } |
| 117 | } |
| 118 | |
| 119 | fn public(ids: &[&str]) -> HashSet<String> { |
| 120 | ids.iter().map(|id| (*id).to_owned()).collect() |
| 121 | } |
| 122 | |
| 123 | #[test] |
| 124 | fn a_public_work_item_changes_every_repository_it_is_linked_to() { |
| 125 | let data = json!({ "itemId": "wki_1", "workspace": "acme", "number": 40, "repoIds": ["rep_1", "rep_2"] }); |
| 126 | let with_origin = event("work_item.updated", Some("rep_1"), data); |
| 127 | let repos: Vec<_> = changes_for(&with_origin, &public(&["rep_1"])).into_iter().map(|change| (change.repo_id, change.number)).collect(); |
| 128 | assert_eq!(repos, vec![("rep_1".to_owned(), Some(40)), ("rep_2".to_owned(), Some(40))]); |
| 129 | assert_eq!(origins_linked(&[with_origin]), vec!["rep_1".to_owned()]); |
| 130 | // What changes no page stays quiet. |
| 131 | assert!(changes_for(&event("audit.recorded", None, json!({ "repoIds": ["rep_1"] })), &public(&["rep_1"])).is_empty()); |
| 132 | } |
| 133 | |
| 134 | #[test] |
| 135 | fn a_private_work_item_changes_only_its_origin() { |
| 136 | let repos = |event: &Event| changes_for(event, &public(&["rep_9"])).into_iter().map(|change| change.repo_id).collect::<Vec<_>>(); |
| 137 | // Its origin is private: the repositories it only names hear nothing. |
| 138 | let data = json!({ "itemId": "wki_1", "workspace": "acme", "number": 40, "repoIds": ["rep_1", "rep_2"] }); |
| 139 | assert_eq!(repos(&event("work_item.updated", Some("rep_1"), data)), vec!["rep_1".to_owned()]); |
| 140 | // Filed in the workspace: no repository hears of it, not even the |
| 141 | // first it is linked to, nor one its data names. |
| 142 | let filed = json!({ "itemId": "wki_1", "workspace": "acme", "number": 40, "repoId": "rep_2", "repoIds": ["rep_2", "rep_9"] }); |
| 143 | assert!(repos(&event("work_item.linked", None, filed.clone())).is_empty()); |
| 144 | assert!(repos(&event("comment.created", None, filed)).is_empty()); |
| 145 | } |
| 146 | } |