johnandersen777.bsky.social/knotless
git that lives in your Atmosphere account
Forked from chadtmiller.com/knotless
0
0
//! `dev.knotless.getFeed`: what happened on the repositories the index
//! knows, newest first. The front page shows it, all of it or what those
//! the viewer follows did.
use super::*;
use crate::index::{Among, Assembled, Index, Push, cursor_at};
use crate::viewer::Viewer;
use axum::Extension;
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;
/// Seconds between pushes to a ref that are shown as one.
const RUN: i64 = 60 * 60;
#[derive(Serialize)]
pub struct Output {
events: Vec<Event>,
/// Where the next page starts, when there may be one.
#[serde(skip_serializing_if = "Option::is_none")]
cursor: Option<String>,
}
/// Something an account did to a repository.
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct Event {
created_at: String,
actor: ActorView,
repo: RepoView,
action: Action,
}
/// A repository, under the name it has now.
#[derive(Serialize, Clone)]
pub struct RepoView {
did: String,
handle: String,
name: String,
}
#[derive(Serialize)]
pub struct CommitView {
sha: String,
message: String,
/// Whether it joins more than one line of history.
#[serde(skip_serializing_if = "std::ops::Not::not")]
merge: bool,
}
#[derive(Serialize)]
#[serde(tag = "$type")]
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")]
description: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
fork_of: Option<RepoView>,
},
/// A ref made or moved, and the newest of the commits that brought
/// which no other ref had first.
#[serde(rename = "dev.knotless.getFeed#pushed")]
Pushed {
#[serde(rename = "ref")]
refname: String,
created: bool,
/// The commit the ref was left at, which outlives the ref.
to: String,
#[serde(skip_serializing_if = "std::ops::Not::not")]
forced: bool,
count: usize,
new: usize,
commits: Vec<CommitView>,
},
#[serde(rename = "dev.knotless.getFeed#deleted")]
Deleted {
#[serde(rename = "ref")]
refname: String,
},
/// A pull or an issue opened.
#[serde(rename = "dev.knotless.getFeed#opened", rename_all = "camelCase")]
Opened {
subject: String,
title: String,
/// Where the subject stands now, whoever moved it there.
#[serde(skip_serializing_if = "Option::is_none")]
subject_state: Option<String>,
},
/// A comment, or a review with a verdict.
#[serde(rename = "dev.knotless.getFeed#commented", rename_all = "camelCase")]
Commented {
subject: String,
title: String,
text: String,
#[serde(skip_serializing_if = "Option::is_none")]
verdict: Option<String>,
/// Where the subject stands now, whoever moved it there.
#[serde(skip_serializing_if = "Option::is_none")]
subject_state: Option<String>,
},
/// A pull or an issue moved to another state.
#[serde(rename = "dev.knotless.getFeed#changed", rename_all = "camelCase")]
Changed {
subject: String,
title: String,
state: String,
/// Where the subject stands now, whoever moved it there.
#[serde(skip_serializing_if = "Option::is_none")]
subject_state: Option<String>,
},
/// The event's repository starred, by its record.
#[serde(rename = "dev.knotless.getFeed#starred")]
Starred { subject: String },
}
#[derive(Deserialize, Default)]
pub struct Params {
limit: Option<usize>,
cursor: Option<String>,
/// Only what this account, by DID, did.
actor: Option<String>,
/// `following` for what the viewer and those it follows did, and
/// anything on its repositories and those it starred; else all.
filter: Option<String>,
/// Only `pushes`, `pulls`, `issues`, `comments`, `stars` or `repos`.
kind: Option<String>,
}
pub async fn get_feed(
State(app): State<Arc<App>>,
Extension(viewer): Extension<Viewer>,
Query(p): Query<Params>,
) -> Result<axum::Json<Output>> {
let index = app.index.as_ref().ok_or(Shown(
StatusCode::NOT_FOUND,
"this server keeps no index".into(),
))?;
let following = match p.filter.as_deref() {
Some("following") => Some(viewer.0.ok_or(Shown(
StatusCode::UNAUTHORIZED,
"log in to see what those you follow did".into(),
))?),
_ => None,
};
let limit = p.limit.unwrap_or(LIMIT).clamp(1, 100);
let whose = Whose {
by: p.actor.as_deref(),
following: following.as_deref(),
kind: p.kind.as_deref(),
};
let page = events(&app, index, limit, p.cursor.as_deref(), whose);
let (events, cursor) = page.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()))
}
/// Whose doings a page holds: `by`'s alone, if named; or `following`'s
/// and those it follows', with anything on its repositories and those it
/// starred, if named; else everyone's but their stars, which anyone can
/// write as many of as they like. Of every kind, or only `kind`: a kind
/// this server does not know is none. Unless `by` is named, of each
/// repository's bots, the runners and triagers it names that are not
/// people who may push, only the pulls they merged, as whoever queued
/// each: what they say and push restates what people did.
#[derive(Clone, Copy)]
struct Whose<'a> {
by: Option<&'a str>,
following: Option<&'a str>,
kind: Option<&'a str>,
}
/// 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, what was opened, said and decided on them, and stars; and where
/// the next page starts. A page is the newest `limit` records of all four
/// kinds, of those `whose` names, by time and then URI, less those not
/// shown.
async fn events(
app: &App,
index: &Index,
limit: usize,
cursor: Option<&str>,
whose: Whose<'_>,
) -> Result<(Vec<Event>, Option<String>)> {
let Whose {
by,
following,
kind,
} = whose;
let wants = |k: &str| kind.is_none_or(|kind| kind == k);
// One account's feed is all it did; the rest leave out bots.
let quiet = by.is_none();
// Pulls and issues are their records and changes of their state.
let only = match kind {
Some("pulls") => Some("pull"),
Some("issues") => Some("issue"),
Some("comments") => Some("comment"),
_ => None,
};
let among = Among {
owner: by,
followed_by: following,
..Among::default()
};
let (mut repos, mut pushes, mut said, mut stars) = (vec![], vec![], vec![], vec![]);
if wants("repos") {
(repos, _) = index
.repos_page("", limit, cursor, among)
.map_err(anyhow::Error::from)
.map_err(AppError::internal)?;
}
if wants("pushes") {
pushes = index
.pushes(limit, cursor, by, following, quiet)
.map_err(anyhow::Error::from)
.map_err(AppError::internal)?;
}
if kind.is_none() || only.is_some() {
said = index
.recent(limit, cursor, by, following, only, quiet)
.map_err(anyhow::Error::from)
.map_err(AppError::internal)?;
}
// Stars only of one account, or of whom one follows.
if wants("stars") && by.or(following).is_some() {
stars = index
.stars(limit, cursor, by, following)
.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())
}))
.chain(
stars
.iter()
.map(|s| (s.created_at.as_str(), s.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));
stars.retain(|s| held(&s.created_at, &s.uri));
}
// Pulls and issues, as their repositories' owners let them stand, by
// title, and the repositories they are on.
let mut titles: HashMap<String, String> = HashMap::new();
let mut states: HashMap<String, String> = HashMap::new();
let on: HashSet<(&str, &str)> = said
.iter()
.map(|(_, did, key)| (did.as_str(), key.as_str()))
.collect();
let mut on_repos: HashMap<(&str, &str), (String, RepoRecord, Arc<Assembled>)> = 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 record = index
.repo_of(did, key)
.map_err(anyhow::Error::from)
.map_err(AppError::internal)?;
let record: RepoRecord = record
.and_then(|v| serde_json::from_value(v).ok())
.unwrap_or_default();
let made = index
.assembled(did, key, &record)
.map_err(anyhow::Error::from)
.map_err(AppError::internal)?;
for e in &made.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.clone(), title);
}
}
states.extend(made.items.iter().map(|i| (i.uri.clone(), i.state.clone())));
on_repos.insert((did, key), (name, record, made));
}
// Who each is shown as said or done by.
let authors: Vec<&str> = said
.iter()
.map(|(e, did, key)| {
let (_, record, made) = &on_repos[&(did.as_str(), key.as_str())];
shown_by(e, did, record, made)
})
.collect();
// 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().flat_map(|p| [p.did.as_str(), p.by.as_str()]));
dids.extend(said.iter().map(|(_, did, _)| did.as_str()));
dids.extend(authors.iter().copied());
dids.extend(
stars
.iter()
.flat_map(|s| [s.did.as_str(), s.owner.as_str()]),
);
let actors: HashMap<&str, ActorView> = futures::future::join_all(
dids.into_iter()
.map(|did| async move { (did, super::actor(app, did).await) }),
)
.await
.into_iter()
.collect();
let actor = |did: &str| actors[did].clone();
let repo = |did: &str, name: &str| RepoView {
did: did.to_string(),
handle: actors[did].handle.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();
// Each commit is listed once, by the earliest run on the page to bring
// it that no other ref of its repository had it before.
let mut listed: HashSet<(&str, String)> = HashSet::new();
for run in runs(&pushes).iter().rev() {
let (last, first) = (run[0], run[run.len() - 1]);
let mirror = mirrors.get(&(last.did.clone(), last.name.clone()));
let action = match &last.to {
None => Action::Deleted {
refname: last.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, &first.created_at);
let known: Vec<String> = match &first.from {
Some(from) => vec![from.clone()],
None => [
(branch != first.refname).then_some(first.repo.as_str()),
first.fork_of.as_deref(),
]
.into_iter()
.flatten()
.map(before)
.collect::<rusqlite::Result<Vec<_>>>()
.map_err(anyhow::Error::from)
.map_err(AppError::internal)?
.into_iter()
.flatten()
.collect(),
};
let others = index
.tips_before(&first.repo, &first.refname, &first.created_at, &first.uri)
.map_err(anyhow::Error::from)
.map_err(AppError::internal)?;
let walk_from = |tip: &str, known: &[&str]| {
mirror
.and_then(|r| r.objects().ok())
.and_then(|o| o.brought(tip, known, WALKED).ok())
.unwrap_or_default()
};
let walk = |known: &[&str]| walk_from(to, known);
// A move that left commits behind was forced, a rebase or
// a rewrite: what it brought of them again, by the same
// author at the same time with the same message, is not
// news.
let replaced = match &first.from {
Some(from) => walk_from(from, &[to.as_str()]),
None => Vec::new(),
};
let again: HashSet<_> = replaced.iter().map(same).collect();
let fresh = |c: &read::Commit| !again.contains(&same(c));
let mut known: Vec<&str> = known.iter().map(String::as_str).collect();
let brought: Vec<_> = walk(&known).into_iter().filter(fresh).collect();
// A fork's copy of its source's refs is the fork itself.
if first.fork_of.is_some() && first.from.is_none() && brought.is_empty() {
continue;
}
known.extend(others.iter().map(String::as_str));
let new: Vec<_> = walk(&known)
.into_iter()
.filter(fresh)
.filter(|c| listed.insert((last.repo.as_str(), c.sha.clone())))
.collect();
Action::Pushed {
refname: last.refname.clone(),
created: first.from.is_none(),
to: to.clone(),
forced: !replaced.is_empty(),
count: brought.len(),
new: new.len(),
commits: new
.into_iter()
.take(LISTED)
.map(|c| CommitView {
merge: c.parents.len() > 1,
sha: c.sha,
message: c.message,
})
.collect(),
}
}
};
let event = Event {
created_at: last.created_at.clone(),
actor: actor(&last.by),
repo: repo(&last.did, &last.name),
action,
};
events.push((last.uri.clone(), event));
}
let named: HashMap<(&str, &str), RepoView> = on_repos
.iter()
.map(|(&(did, key), (name, ..))| ((did, key), repo(did, name)))
.collect();
for ((e, did, key), author) in said.iter().zip(&authors) {
// 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 subject_state = states.get(subject).cloned();
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,
subject_state,
},
nsid::COMMENT => Action::Commented {
subject,
title,
text: text("body"),
verdict: e.value["verdict"].as_str().map(str::to_string),
subject_state,
},
_ => Action::Changed {
subject,
title,
state: text("state"),
subject_state,
},
};
let event = Event {
created_at: text("createdAt"),
actor: actor(author),
repo: named[&(did.as_str(), key.as_str())].clone(),
action,
};
events.extend(shown.then(|| (e.uri.clone(), event)));
}
for s in &stars {
let event = Event {
created_at: s.created_at.clone(),
actor: actor(&s.did),
repo: repo(&s.owner, &s.name),
action: Action::Starred {
subject: s.repo.clone(),
},
};
events.push((s.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))
}
/// Who `e`, on `owner`'s repository of `record`, is shown as written by:
/// its author, or, for a pull a runner the record names marked merged
/// landing the merge queue, whoever queued it, as the pull's standing in
/// the queue says. A runner that is the owner or a collaborator is a
/// person, and keeps it.
fn shown_by<'a>(
e: &'a forge::Entry,
owner: &str,
record: &RepoRecord,
made: &'a Assembled,
) -> &'a str {
let person = e.author == owner || record.is_collaborator(&e.author);
let bot = record.is_runner(&e.author) && !person;
let merged = e.value["state"] == "merged" && bot;
let subject = e.value["subject"].as_str().unwrap_or_default();
let item = made.items.iter().find(|i| merged && i.uri == subject);
let queued = item
.and_then(|i| i.queue.as_ref())
.filter(|q| q.state == "queued");
queued.map_or(&e.author, |q| &q.by)
}
/// What a rebase or a rewrite keeps of a commit: who wrote it, when, and
/// what it says.
fn same(c: &read::Commit) -> (&str, &str, &str, &str) {
(&c.author.name, &c.author.email, &c.author.date, &c.message)
}
/// `pushes`, newest first, in runs shown as one, newest first, each newest
/// first: one account's pushes to a ref, each within an hour of the one
/// after it.
/// A deletion is a run of its own, which ends the run after it.
fn runs(pushes: &[Push]) -> Vec<Vec<&Push>> {
let mut runs: Vec<Vec<&Push>> = Vec::new();
let mut open: HashMap<(&str, &str, &str), usize> = HashMap::new();
for p in pushes {
let key = (p.repo.as_str(), p.refname.as_str(), p.by.as_str());
let after = open.get(&key).filter(|&&i| {
let next = &runs[i][runs[i].len() - 1].created_at;
p.to.is_some() && within(&p.created_at, next)
});
match after {
Some(&i) => runs[i].push(p),
None if p.to.is_none() => {
open.remove(&key);
runs.push(vec![p]);
}
None => {
open.insert(key, runs.len());
runs.push(vec![p]);
}
}
}
runs
}
/// Whether `next` came within a run's gap of `at`.
fn within(at: &str, next: &str) -> bool {
seconds(at)
.zip(seconds(next))
.is_some_and(|(at, next)| next - at <= RUN)
}
/// Seconds since 1970 at `at`, a UTC datetime such as
/// `2026-10-21T22:50:00Z`, by Howard Hinnant's days from civil.
fn seconds(at: &str) -> Option<i64> {
let n = |from: usize, to: usize| at.get(from..to)?.parse::<i64>().ok();
let (y, m, d) = (n(0, 4)?, n(5, 7)?, n(8, 10)?);
let (h, min, s) = (n(11, 13)?, n(14, 16)?, n(17, 19)?);
let y = if m <= 2 { y - 1 } else { y };
let era = y.div_euclid(400);
let yoe = y - era * 400;
let doy = (153 * ((m + 9) % 12) + 2) / 5 + d - 1;
let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy;
Some((era * 146097 + doe - 719468) * 86400 + h * 3600 + min * 60 + s)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn times() {
assert_eq!(seconds("1970-01-01T00:00:00Z"), Some(0));
assert_eq!(seconds("2026-10-21T22:50:07.123Z"), Some(1792623007));
assert_eq!(seconds("2000-02-29T00:00:00Z"), Some(951782400));
assert_eq!(seconds("t"), None);
assert!(within("2026-01-01T00:00:00Z", "2026-01-01T01:00:00Z"));
assert!(!within("2026-01-01T00:00:00Z", "2026-01-01T01:00:01Z"));
assert!(!within("t", "2026-01-01T01:00:00Z"));
}
#[test]
fn pushes_in_runs() {
let by = |by: &str, r: &str, at: &str, to: Option<&str>| Push {
uri: format!("at://did:a/dev.knotless.claim/{at}"),
repo: "at://did:a/dev.knotless.repo/r".into(),
did: "did:a".into(),
name: "r".into(),
fork_of: None,
refname: format!("refs/heads/{r}"),
from: None,
to: to.map(str::to_string),
created_at: format!("2026-01-01T{at}:00Z"),
by: by.into(),
};
let push = |r: &str, at: &str, to: Option<&str>| by("did:a", r, at, to);
// Newest first: main moved at five, four and three, a gap, then at
// one; side made at three thirty; main deleted at half past four;
// and a collaborator moved main at two, between the owner's moves.
let pushes = [
push("main", "05:00", Some("e")),
push("main", "04:30", None),
push("main", "04:00", Some("d")),
push("side", "03:30", Some("s")),
push("main", "03:00", Some("c")),
by("did:b", "main", "02:00", Some("b")),
push("main", "01:00", Some("a")),
];
let runs: Vec<Vec<&str>> = runs(&pushes)
.into_iter()
.map(|run| run.into_iter().map(|p| &p.created_at[11..16]).collect())
.collect();
assert_eq!(
runs,
[
vec!["05:00"],
vec!["04:30"],
vec!["04:00", "03:00"],
vec!["03:30"],
vec!["02:00"],
vec!["01:00"]
]
);
}
#[test]
fn merges_shown_as_queued() {
let pull = "at://did:o/dev.knotless.pull/p";
let queued = |state: &str| knotless::forge::Queued {
state: state.into(),
reason: None,
commit: None,
checks: vec![],
by: "did:o".into(),
at: "q".into(),
created_at: "t".into(),
};
let made = |queue| Assembled {
entries: vec![],
items: vec![knotless::forge::Item {
uri: pull.into(),
queue,
..Default::default()
}],
};
let state = |author: &str, state: &str| forge::Entry {
uri: format!("at://{author}/dev.knotless.state/s"),
author: author.into(),
value: serde_json::json!({"subject": pull, "state": state}),
};
let record: RepoRecord = serde_json::from_value(serde_json::json!({
"name": "r", "key": "k", "createdAt": "t",
"collaborators": [{"did": "did:c"}],
"runners": [{"did": "did:bot"}, {"did": "did:o"}, {"did": "did:c"}],
}))
.unwrap();
let shown = |author: &str, s: &str, made: &Assembled| {
shown_by(&state(author, s), "did:o", &record, made).to_string()
};
let queue = made(Some(queued("queued")));
// A runner's merge is whoever queued the pull's.
assert_eq!(shown("did:bot", "merged", &queue), "did:o");
// Anyone else's, a runner's that is the owner or a collaborator,
// and a runner's other changes, their own.
for who in ["did:m", "did:o", "did:c"] {
assert_eq!(shown(who, "merged", &queue), who);
}
assert_eq!(shown("did:bot", "closed", &queue), "did:bot");
// A pull no one queued, or that left the queue, is the runner's.
for made in [made(None), made(Some(queued("left")))] {
assert_eq!(shown("did:bot", "merged", &made), "did:bot");
}
}
#[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);
}
}