//! `dev.knotless.getFeed`: what happened on the repositories the index //! knows, newest first. The front page shows it. use super::*; use crate::index::{Index, cursor_at}; use knotless::records::nsid; use std::collections::HashSet; /// Events answered when the asker names no number. const LIMIT: usize = 40; /// Commits a push is walked for: past this, it says only that there were /// at least as many. const WALKED: usize = 100; /// Commits a push lists; the rest are counted. const LISTED: usize = 10; #[derive(Serialize, TS)] #[ts(export, rename = "Feed")] pub struct Output { events: Vec, /// Where the next page starts, when there may be one. #[serde(skip_serializing_if = "Option::is_none")] #[ts(optional)] cursor: Option, } /// Something an account did to a repository. #[derive(Serialize, TS)] #[serde(rename_all = "camelCase")] #[ts(export, rename = "FeedEvent")] pub struct Event { created_at: String, actor: Actor, repo: RepoView, action: Action, } #[derive(Serialize, TS, Clone)] #[ts(export, rename = "FeedActor")] pub struct Actor { did: String, handle: String, } /// A repository, under the name it has now. #[derive(Serialize, TS, Clone)] #[ts(export, rename = "FeedRepo")] pub struct RepoView { did: String, handle: String, name: String, } #[derive(Serialize, TS)] #[ts(export, rename = "FeedCommit")] pub struct CommitView { sha: String, message: String, } #[derive(Serialize, TS)] #[serde(tag = "$type")] #[ts(export, rename = "FeedAction")] pub enum Action { /// A repository made, or forked from another. #[serde(rename = "dev.knotless.getFeed#created", rename_all = "camelCase")] Created { #[serde(skip_serializing_if = "Option::is_none")] #[ts(optional)] description: Option, #[serde(skip_serializing_if = "Option::is_none")] #[ts(optional)] fork_of: Option, }, /// A ref made or moved, and the newest of the commits that brought. #[serde(rename = "dev.knotless.getFeed#pushed")] Pushed { #[serde(rename = "ref")] refname: String, created: bool, count: usize, commits: Vec, }, #[serde(rename = "dev.knotless.getFeed#deleted")] Deleted { #[serde(rename = "ref")] refname: String, }, /// A pull or an issue opened. #[serde(rename = "dev.knotless.getFeed#opened")] Opened { subject: String, title: String }, /// A comment, or a review with a verdict. #[serde(rename = "dev.knotless.getFeed#commented")] Commented { subject: String, title: String, text: String, #[serde(skip_serializing_if = "Option::is_none")] #[ts(optional)] verdict: Option, }, /// A pull or an issue moved to another state. #[serde(rename = "dev.knotless.getFeed#changed")] Changed { subject: String, title: String, state: String, }, } #[derive(Deserialize, Default)] pub struct Params { limit: Option, cursor: Option, } pub async fn get_feed( State(app): State>, Query(p): Query, ) -> Result> { let index = app.index.as_ref().ok_or(Shown( StatusCode::NOT_FOUND, "this server keeps no index".into(), ))?; let limit = p.limit.unwrap_or(LIMIT).clamp(1, 100); let (events, cursor) = events(&app, index, limit, p.cursor.as_deref()).await?; Ok(axum::Json(Output { events, cursor })) } /// The owner and name of the repository `uri` names. fn owner_and_name(uri: &str) -> Option<(String, String)> { let (did, _, name) = forge::parse_uri(uri).ok()?; Some((did.to_string(), name.to_string())) } /// Where a page of `limit` ends, from the time and URI of each record it /// could hold, newest `limit` of each kind: the last it holds, if it is /// full. fn cut<'a>(mut keys: Vec<(&'a str, &'a str)>, limit: usize) -> Option<(&'a str, &'a str)> { keys.sort_by(|a, b| b.cmp(a)); keys.get(limit - 1).copied() } /// What happened, newest first, after `cursor`: repositories made, refs /// moved, and what was opened, said and decided on them; and where the /// next page starts. A page is the newest `limit` records of all three /// kinds, by time and then URI, less those not shown. async fn events( app: &App, index: &Index, limit: usize, cursor: Option<&str>, ) -> Result<(Vec, Option)> { let (mut repos, _) = index .repos_page("", limit, cursor) .map_err(anyhow::Error::from) .map_err(AppError::internal)?; let mut pushes = index .pushes(limit, cursor) .map_err(anyhow::Error::from) .map_err(AppError::internal)?; let mut said = index .recent(limit, cursor) .map_err(anyhow::Error::from) .map_err(AppError::internal)?; let at = |e: &forge::Entry| { e.value["createdAt"] .as_str() .unwrap_or_default() .to_string() }; let keys = repos .iter() .map(|r| (r.created_at.as_str(), r.uri.as_str())) .chain( pushes .iter() .map(|p| (p.created_at.as_str(), p.uri.as_str())), ) .chain(said.iter().map(|(e, _, _)| { let at = e.value["createdAt"].as_str().unwrap_or_default(); (at, e.uri.as_str()) })) .collect(); let end = cut(keys, limit).map(|(at, uri)| (at.to_string(), uri.to_string())); let next = end.as_ref().map(|(at, uri)| cursor_at(at, uri)); if let Some(end) = &end { let held = |at: &str, uri: &str| (at, uri) >= (end.0.as_str(), end.1.as_str()); repos.retain(|r| held(&r.created_at, &r.uri)); pushes.retain(|p| held(&p.created_at, &p.uri)); said.retain(|(e, _, _)| held(&at(e), &e.uri)); } // Each account named, resolved together. let parents: Vec<(String, String)> = repos .iter() .filter_map(|r| r.fork_of.as_deref()) .filter_map(owner_and_name) .collect(); let mut dids: HashSet<&str> = parents.iter().map(|(did, _)| did.as_str()).collect(); dids.extend(repos.iter().map(|r| r.did.as_str())); dids.extend(pushes.iter().map(|p| p.did.as_str())); dids.extend( said.iter() .flat_map(|(e, did, _)| [e.author.as_str(), did.as_str()]), ); let handles: HashMap<&str, String> = futures::future::join_all( dids.into_iter() .map(|did| async move { (did, handle(app, did).await) }), ) .await .into_iter() .collect(); let actor = |did: &str| Actor { did: did.to_string(), handle: handles[did].clone(), }; let repo = |did: &str, name: &str| RepoView { did: did.to_string(), handle: handles[did].clone(), name: name.to_string(), }; let mut events: Vec<(String, Event)> = repos .iter() .map(|r| { ( r.uri.clone(), Event { created_at: r.created_at.clone(), actor: actor(&r.did), repo: repo(&r.did, &r.name), action: Action::Created { description: r.description.clone(), fork_of: r .fork_of .as_deref() .and_then(owner_and_name) .map(|(did, name)| repo(&did, &name)), }, }, ) }) .collect(); // The mirror of each repository pushed to, opened once. let wanted: HashSet<(String, String)> = pushes .iter() .map(|p| (p.did.clone(), p.name.clone())) .collect(); let mirrors: HashMap<(String, String), Repo> = futures::future::join_all(wanted.into_iter().map(|at| async move { let repo = Repo::open(app, &at.0, &at.1).await; (at, repo) })) .await .into_iter() .filter_map(|(at, repo)| Some((at, repo.ok()?))) .collect(); for p in &pushes { let mirror = mirrors.get(&(p.did.clone(), p.name.clone())); let action = match &p.to { None => Action::Deleted { refname: p.refname.clone(), }, Some(to) => { // A ref that moved brought what it did not reach before; a // new one, what the default branch did not reach just // before, nor the default branch of a fork's source, which // a fork's is named for. let head = mirror.map(|r| r.head.as_str()).unwrap_or_default(); let branch = format!("refs/heads/{head}"); let before = |at: &str| index.tip_before(at, &branch, &p.created_at); let known: Vec = match &p.from { Some(from) => vec![from.clone()], None => [ (branch != p.refname).then_some(p.repo.as_str()), p.fork_of.as_deref(), ] .into_iter() .flatten() .map(before) .collect::>>() .map_err(anyhow::Error::from) .map_err(AppError::internal)? .into_iter() .flatten() .collect(), }; let known: Vec<&str> = known.iter().map(String::as_str).collect(); let brought = mirror .and_then(|r| r.objects().ok()) .and_then(|o| o.brought(to, &known, WALKED).ok()) .unwrap_or_default(); // A fork's copy of its source's refs is the fork itself. if p.fork_of.is_some() && p.from.is_none() && brought.is_empty() { continue; } Action::Pushed { refname: p.refname.clone(), created: p.from.is_none(), count: brought.len(), commits: brought .into_iter() .take(LISTED) .map(|c| CommitView { sha: c.sha, message: c.message, }) .collect(), } } }; let event = Event { created_at: p.created_at.clone(), actor: actor(&p.did), repo: repo(&p.did, &p.name), action, }; events.push((p.uri.clone(), event)); } // Pulls and issues, as their repositories' owners let them stand, by // title. let mut titles: HashMap = HashMap::new(); let on: HashSet<(&str, &str)> = said .iter() .map(|(_, did, key)| (did.as_str(), key.as_str())) .collect(); let mut named = HashMap::new(); for (did, key) in on { let name = index .name(did, key) .map_err(anyhow::Error::from) .map_err(AppError::internal)? .unwrap_or_default(); let entries = index .forge(did, key) .map_err(anyhow::Error::from) .map_err(AppError::internal)?; for e in entries { let collection = forge::parse_uri(&e.uri).unwrap_or_default().1; if matches!(collection, nsid::PULL | nsid::ISSUE) { let title = e.value["title"].as_str().unwrap_or_default().to_string(); titles.insert(e.uri, title); } } named.insert((did, key), repo(did, &name)); } for (e, did, key) in &said { // A pull or an issue is its own subject. One the repository's owner // keeps out is not shown, nor what is said about it. let subject = if titles.contains_key(&e.uri) { e.uri.as_str() } else { e.value["subject"].as_str().unwrap_or_default() }; let shown = titles.contains_key(subject); let title = titles.get(subject).cloned().unwrap_or_default(); let text = |field: &str| e.value[field].as_str().unwrap_or_default().to_string(); let subject = subject.to_string(); let action = match forge::parse_uri(&e.uri).unwrap_or_default().1 { nsid::PULL | nsid::ISSUE => Action::Opened { subject, title }, nsid::COMMENT => Action::Commented { subject, title, text: text("body"), verdict: e.value["verdict"].as_str().map(str::to_string), }, _ => Action::Changed { subject, title, state: text("state"), }, }; let event = Event { created_at: text("createdAt"), actor: actor(&e.author), repo: named[&(did.as_str(), key.as_str())].clone(), action, }; events.extend(shown.then(|| (e.uri.clone(), event))); } events.sort_by(|(a, x), (b, y)| (&y.created_at, b).cmp(&(&x.created_at, a))); Ok((events.into_iter().map(|(_, e)| e).collect(), next)) } #[cfg(test)] mod tests { use super::*; #[test] fn owners_and_names() { assert_eq!( owner_and_name("at://did:a/dev.knotless.repo/r"), Some(("did:a".into(), "r".into())) ); assert_eq!(owner_and_name("nope"), None); } }