johnandersen777.bsky.social/knotless
git that lives in your Atmosphere account
Forked from chadtmiller.com/knotless
0
0
//! A profile's badges, `dev.knotless.getProfile#badgeView`: counted from
//! the index as the profile is asked for, and from where the account
//! lives, as its DID document and PDS say, looked up now and then. None is
//! stored in anyone's PDS, so none can be claimed by writing a record, and
//! every one comes back whenever the index is rebuilt and refilled.
use super::*;
use crate::index::{Home, Pulls};
use knotless::records::queue::{CONFLICTS, EJECTED, HALVED};
/// `dev.knotless.getProfile#badgeView`.
#[derive(Serialize, Debug, PartialEq)]
pub struct BadgeView {
kind: &'static str,
tier: usize,
count: u64,
#[serde(skip_serializing_if = "Option::is_none")]
next: Option<u64>,
}
/// A badge earned once: its one tier.
const ONCE: &[u64] = &[1];
/// Each badge, in the order a profile shows them, and the counts its
/// tiers start at: by group, collaboration, the merge queue, packages and
/// stars, atproto, and those that can no longer be earned.
const BADGES: [(&str, &[u64]); 12] = [
("upstream", &[2, 16, 128, 1024]),
("second-pair", &[1, 10, 50, 250]),
("peacemaker", &[1, 5, 25, 100]),
("forked-and-fixed", ONCE),
("good-company", &[1, 10, 50, 200]),
("smooth-landing", &[1, 10, 50, 250]),
("bisected", ONCE),
("shipped", &[1, 10, 50, 200]),
("stargazer", &[16, 128, 512, 4096]),
("carry-your-own", ONCE),
("moved-house", ONCE),
("first-light", ONCE),
];
/// How many pulls a batch of the merge queue lands together for good
/// company.
const COMPANY: usize = 5;
/// The most accounts a PDS may hold and still be its account's own.
const OWN_PDS_HOLDS: usize = 5;
/// How long what was found of where an account lives is kept before it is
/// looked up again.
const HOME_KEPT_FOR: Duration = Duration::from_secs(24 * 60 * 60);
/// The accounts here before knotless 1.0, a DID a line, as the list stood
/// that day. Until it is written, every account with a repository is.
const FIRST_LIGHT: &str = include_str!("first_light.txt");
/// The badges `counts` earn, one for each of [`BADGES`] in turn: each at
/// the highest tier its count reaches, and none that reaches no tier.
fn badges(counts: [u64; 12]) -> Vec<BadgeView> {
let earned = BADGES.iter().zip(counts).map(|((kind, at), count)| {
let tier = at.iter().filter(|&&n| count >= n).count();
let next = at.get(tier).copied();
(tier > 0).then_some(BadgeView {
kind,
tier,
count,
next,
})
});
earned.flatten().collect()
}
/// The badges `did` earned, counted from the index, and from `home`,
/// where its account lives.
pub(super) fn badges_of(index: &Index, did: &str, home: Home) -> Vec<BadgeView> {
let (shipped, stars, repos) = index.badge_counts(did).unwrap_or_default();
let mut c = Tally::default();
for pulls in index.pulls_by(did).unwrap_or_default() {
let items = forge::assemble(&pulls.repo, &pulls.owner, &pulls.entries);
let ejected = ejections(&pulls);
// How many each batch landed, read once the first merged pull
// names one: a query for each batch read the repository's records
// as many times over.
let mut batches = None;
for item in items.iter().filter(|i| i.author == did) {
let ejected = ejected
.get(item.uri.as_str())
.map_or(&[][..], Vec::as_slice);
c.authored(&pulls.owner, item, ejected, || {
batches.get_or_insert_with(|| landed(index, &pulls))
});
}
}
let first = index.first_comments(did).unwrap_or_default();
for pulls in index.pulls_commented(did).unwrap_or_default() {
let items = forge::assemble(&pulls.repo, &pulls.owner, &pulls.entries);
let merged = items.iter().filter(|i| i.state == "merged");
let before = |i: &&Item| {
first
.get(&i.uri)
.is_some_and(|f| reviewed(&pulls, i, f, did))
};
c.second_pair += merged.filter(before).count() as u64;
}
for pulls in index.pulls_from_forks(did).unwrap_or_default() {
let items = forge::assemble(&pulls.repo, &pulls.owner, &pulls.entries);
c.forked += items.iter().filter(|i| i.state == "merged").count() as u64;
}
badges([
c.upstream,
c.second_pair,
c.peacemaker,
c.forked,
c.company,
c.smooth,
c.bisected,
shipped,
stars,
home.own as u64,
home.moved as u64,
first_light(FIRST_LIGHT, did, repos) as u64,
])
}
/// What the pulls an account took part in count toward its badges.
#[derive(Default)]
struct Tally {
upstream: u64,
second_pair: u64,
peacemaker: u64,
forked: u64,
company: u64,
smooth: u64,
bisected: u64,
}
impl Tally {
/// Counts `item`, a pull the account opened on the repository `owner`
/// keeps: merged into another's repository; landed by the merge
/// queue, never turned away, or in a batch of many; turned away for a
/// conflict and landed after all; or found by halving a batch.
/// `ejected` says why the runner turned it away each time, and
/// `batches` how many pulls each batch of its repository landed.
fn authored<'a>(
&mut self,
owner: &str,
item: &Item,
ejected: &[&str],
batches: impl FnOnce() -> &'a HashMap<String, usize>,
) {
let merged = item.state == "merged";
self.upstream += (merged && owner != item.author) as u64;
let conflicted = ejected.iter().any(|r| r.contains(CONFLICTS));
self.peacemaker += (merged && conflicted) as u64;
self.bisected += ejected.iter().any(|r| r.contains(HALVED)) as u64;
let Some(batch) = item.merged_batch.as_ref().filter(|_| merged) else {
return;
};
self.smooth += ejected.is_empty() as u64;
let landed = batches().get(batch).copied().unwrap_or_default();
self.company += (landed >= COMPANY) as u64;
}
}
/// Why the runner of the repository of `pulls` turned each pull away from
/// its merge queue, each time it did, by the pull's URI.
fn ejections(pulls: &Pulls) -> HashMap<&str, Vec<&str>> {
let mut ejected: HashMap<&str, Vec<&str>> = HashMap::new();
let queue = pulls
.entries
.iter()
.filter(|e| e.uri.contains("/dev.knotless.queue/"));
for e in queue {
let turned = e.value["state"] == EJECTED && pulls.repo.is_runner(&e.author);
if let Some(subject) = e.value["subject"].as_str().filter(|_| turned) {
let reason = e.value["reason"].as_str().unwrap_or_default();
ejected.entry(subject).or_default().push(reason);
}
}
ejected
}
/// How many pulls the merge queue landed in each batch on the repository
/// of `pulls`, as those who may merge said, by the batch's tip.
fn landed(index: &Index, pulls: &Pulls) -> HashMap<String, usize> {
let states = index.batch_states(&pulls.owner, &pulls.repo.key);
let states = states.unwrap_or_default();
let merged = states.iter().filter(|e| {
let at = e.uri.rsplit('/').next().unwrap_or_default();
let by = forge::allowed(&pulls.repo, &pulls.owner, "", &e.author, "merged", at);
e.value["state"] == "merged" && by
});
let mut batches: HashMap<&str, HashSet<&str>> = HashMap::new();
for e in merged {
if let (Some(batch), Some(pull)) = (e.value["batch"].as_str(), e.value["subject"].as_str())
{
batches.entry(batch).or_default().insert(pull);
}
}
let counted = batches
.into_iter()
.map(|(b, pulls)| (b.to_string(), pulls.len()));
counted.collect()
}
/// Whether `did` first commented on `item`, at `first`, before it merged:
/// before the first time anyone else said it merged.
fn reviewed(pulls: &Pulls, item: &Item, first: &str, did: &str) -> bool {
let merges = pulls.entries.iter().filter(|e| {
e.uri.contains("/dev.knotless.state/")
&& e.value["subject"] == item.uri.as_str()
&& e.value["state"] == "merged"
&& e.author != did
});
let merged = merges
.filter_map(|e| when(e.value["createdAt"].as_str()?))
.min();
let first = when(first);
first
.zip(merged)
.is_some_and(|(first, merged)| first < merged)
}
/// A record's time, as it says it.
fn when(at: &str) -> Option<jiff::Timestamp> {
at.parse().ok()
}
/// Whether `did` was here before knotless 1.0: as `list` says, once it is
/// written, and until then, whether it has any of `repos`.
fn first_light(list: &str, did: &str, repos: u64) -> bool {
let listed: Vec<&str> = list
.lines()
.map(str::trim)
.filter(|l| !l.is_empty())
.collect();
let unwritten = listed.is_empty() && repos > 0;
unwritten || listed.contains(&did)
}
/// Where `remote`'s account lives, as its badges count it: as `index` last
/// found it, and looked up again once that is a day old. One never looked
/// up is looked up now.
pub(super) async fn home(app: &App, index: Arc<Index>, remote: &Remote) -> Home {
let (did, pds) = (remote.did().to_string(), remote.identity.pds.clone());
let Some((home, checked)) = index.home(&did).ok().flatten() else {
return look(app.ctx.clone(), index, did, pds).await;
};
if now_secs() - checked >= HOME_KEPT_FOR.as_secs() as i64 {
// Marked as looked up now, so the pages showing it meanwhile look
// it up once between them.
let _ = index.set_home(&did, home, now_secs());
tokio::spawn(look(app.ctx.clone(), index, did, pds));
}
home
}
/// Looks up now where `did`'s account lives, at `pds`, and keeps what was
/// found in `index`. Neither its DID's history nor its PDS answering, it
/// keeps nothing, to be looked up again.
async fn look(
ctx: knotless::env::Ctx,
index: Arc<crate::index::Index>,
did: String,
pds: String,
) -> Home {
let (history, holds) = tokio::join!(history(&ctx, &did), holds(&ctx, &pds));
let answered = history.is_ok() || holds.is_ok();
let home = Home {
moved: history.is_ok_and(|log| moved(&log)),
own: holds.is_ok_and(|repos| own(&repos, &did)),
};
let _ = answered.then(|| index.set_home(&did, home, now_secs()));
home
}
/// The history of `did`'s DID document, as its directory keeps it: only a
/// did:plc has one.
async fn history(ctx: &knotless::env::Ctx, did: &str) -> anyhow::Result<serde_json::Value> {
anyhow::ensure!(did.starts_with("did:plc:"), "{did} keeps no history");
let url = format!("{}/{did}/log/audit", ctx.env.plc_url);
let res = ctx.http.get(url).send().await?.error_for_status()?;
Ok(res.json().await?)
}
/// Whether the history `log` of a DID document named more than one PDS:
/// the account moved house.
fn moved(log: &serde_json::Value) -> bool {
let ops = log.as_array().into_iter().flatten();
let ops = ops
.filter(|e| e["nullified"] != true)
.map(|e| &e["operation"]);
let named = ops.filter_map(|op| {
let pds = op["services"]["atproto_pds"]["endpoint"].as_str();
pds.or(op["service"].as_str())
});
let homes: HashSet<&str> = named.map(|p| p.trim_end_matches('/')).collect();
homes.len() > 1
}
/// The first few accounts `pds` holds, as it lists them: one more than an
/// account's own may hold.
async fn holds(ctx: &knotless::env::Ctx, pds: &str) -> anyhow::Result<Vec<String>> {
#[derive(Deserialize)]
struct Repos {
repos: Vec<Listed>,
}
#[derive(Deserialize)]
struct Listed {
did: String,
}
let xrpc = knotless::xrpc::Xrpc::new(ctx.http.clone(), pds);
let limit = (OWN_PDS_HOLDS + 1).to_string();
let params = [("limit", limit.as_str())];
let listed: Repos = xrpc
.query("com.atproto.sync.listRepos", ¶ms, None)
.await?;
Ok(listed.repos.into_iter().map(|r| r.did).collect())
}
/// Whether a PDS holding `repos` is `did`'s own: it holds `did`, and only a
/// few accounts besides.
fn own(repos: &[String], did: &str) -> bool {
repos.len() <= OWN_PDS_HOLDS && repos.iter().any(|r| r == did)
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
#[test]
fn badges_take_the_highest_tier_reached() {
let shown = |counts| {
badges(counts)
.into_iter()
.map(|b| (b.kind, b.tier, b.next))
.collect::<Vec<_>>()
};
assert_eq!(shown([1, 0, 0, 0, 0, 0, 0, 0, 15, 0, 0, 0]), []);
assert_eq!(
shown([2, 0, 0, 3, 0, 0, 0, 10, 4096, 0, 0, 1]),
[
("upstream", 1, Some(16)),
("forked-and-fixed", 1, None),
("shipped", 2, Some(50)),
("stargazer", 4, None),
("first-light", 1, None)
]
);
}
#[test]
fn first_light_is_every_account_with_a_repository_until_the_list() {
assert!(first_light("", "did:a", 1));
assert!(!first_light("\n", "did:a", 0));
let list = "did:a\n did:b \n";
assert!(first_light(list, "did:b", 0));
assert!(!first_light(list, "did:c", 3));
}
#[test]
fn moved_house_is_a_history_naming_two_pdses() {
let op = |pds: &str| json!({"operation": {"services": {"atproto_pds": {"endpoint": pds}}}});
let legacy = json!({"operation": {"type": "create", "service": "https://a.example/"}});
assert!(!moved(&json!([legacy, op("https://a.example")])));
assert!(moved(&json!([legacy, op("https://b.example")])));
// An operation undone does not count.
let undone = json!({"nullified": true, "operation": {"services": {"atproto_pds": {"endpoint": "https://b.example"}}}});
assert!(!moved(&json!([op("https://a.example"), undone])));
assert!(!moved(&json!({"not": "a log"})));
}
#[test]
fn carry_your_own_is_a_pds_of_a_few_accounts() {
let repos = |n: usize| (0..n).map(|i| format!("did:{i}")).collect::<Vec<_>>();
assert!(own(&repos(1), "did:0"));
assert!(own(&repos(5), "did:4"));
assert!(!own(&repos(6), "did:0"));
assert!(!own(&repos(2), "did:x"));
}
#[tokio::test]
async fn a_home_not_found_is_kept_for_no_one() {
let dir = tempfile::tempdir().unwrap();
let index = Arc::new(Index::open(&dir.path().join("i.db")).unwrap());
let nowhere = "http://127.0.0.1:1";
let env =
knotless::env::Env::from_vars(|k| (k == "KNOTLESS_PLC_URL").then(|| nowhere.into()));
let ctx = knotless::env::Ctx::new(env);
for did in ["did:plc:a", "did:web:a.example"] {
let home = look(ctx.clone(), index.clone(), did.into(), nowhere.into()).await;
assert_eq!(home, Home::default());
assert_eq!(index.home(did).unwrap(), None);
}
}
/// An index with did:a's repository `r`, whose runner is did:ci.
fn index() -> (tempfile::TempDir, Index, serde_json::Value) {
let dir = tempfile::tempdir().unwrap();
let index = Index::open(&dir.path().join("i.db")).unwrap();
let owned = json!({"name": "r", "key": "k", "createdAt": "2026-01-01T00:00:00Z",
"runners": [{"did": "did:ci"}]});
index.put("did:a", nsid::REPO, "r", &owned).unwrap();
let repo = json!({"uri": "at://did:a/dev.knotless.repo/r", "key": "k"});
(dir, index, repo)
}
fn pull(index: &Index, who: &str, n: &str, head: &serde_json::Value) {
let repo = json!({"uri": "at://did:a/dev.knotless.repo/r", "key": "k"});
let pull = json!({"repo": repo, "title": n, "base": "main",
"head": {"repo": head, "branch": "topic"}, "createdAt": "2026-01-02T00:00:00Z"});
index.put(who, nsid::PULL, n, &pull).unwrap();
}
fn state(index: &Index, who: &str, rkey: &str, subject: &str, extra: serde_json::Value) {
let repo = json!({"uri": "at://did:a/dev.knotless.repo/r", "key": "k"});
let mut s = json!({"repo": repo, "state": "merged", "commit": "c",
"subject": subject, "createdAt": "2026-01-03T00:00:00Z"});
s.as_object_mut()
.unwrap()
.extend(extra.as_object().unwrap().clone());
index.put(who, nsid::STATE, rkey, &s).unwrap();
}
fn ids(index: &Index, did: &str) -> Vec<(&'static str, u64)> {
badges_of(index, did, Home::default())
.iter()
.map(|b| (b.kind, b.count))
.collect()
}
#[test]
fn badges_count_pulls_merged_elsewhere() {
let (_dir, index, repo) = index();
for n in ["1", "2", "3"] {
pull(&index, "did:b", n, &repo);
}
let uri = |n: &str| format!("at://did:b/dev.knotless.pull/{n}");
state(&index, "did:a", "3kaaaaaaaaaa2", &uri("1"), json!({}));
state(&index, "did:a", "3kaaaaaaaaaa3", &uri("2"), json!({}));
// Only the owner's word merges: the author's own does not.
state(&index, "did:b", "3kaaaaaaaaaa4", &uri("3"), json!({}));
assert_eq!(ids(&index, "did:b"), [("upstream", 2)]);
// The owner's own pulls are not upstream of anyone; its repository
// is here before 1.0.
assert_eq!(ids(&index, "did:a"), [("first-light", 1)]);
// Releases, and stars on its most starred repository, each
// account's once.
let release = json!({"repo": "k", "package": "p", "version": "1.0.0", "files": [],
"createdAt": "2026-01-01T00:00:00Z"});
index.put("did:a", nsid::RELEASE, "v", &release).unwrap();
let star = json!({"subject": repo, "createdAt": "2026-01-04T00:00:00Z"});
for (who, rkey) in [("did:b", "1"), ("did:b", "2"), ("did:c", "1")] {
index.put(who, nsid::STAR, rkey, &star).unwrap();
}
assert_eq!(index.badge_counts("did:a").unwrap(), (1, 2, 1));
assert_eq!(ids(&index, "did:a"), [("shipped", 1), ("first-light", 1)]);
let home = Home {
moved: true,
own: true,
};
let kinds: Vec<_> = badges_of(&index, "did:c", home)
.iter()
.map(|b| b.kind)
.collect();
assert_eq!(kinds, ["carry-your-own", "moved-house"]);
}
#[test]
fn badges_count_what_the_merge_queue_did() {
let (_dir, index, repo) = index();
let uri = |n: &str| format!("at://did:a/dev.knotless.pull/{n}");
let queue = |who: &str, rkey: &str, n: &str, reason: &str| {
let q = json!({"repo": repo, "subject": uri(n), "state": "ejected",
"reason": reason, "createdAt": "2026-01-02T12:00:00Z"});
index.put(who, nsid::QUEUE, rkey, &q).unwrap();
};
// Pulls 1 to 5 land in one batch, 6 alone; 7 is turned away.
for n in ["1", "2", "3", "4", "5", "6", "7"] {
pull(&index, "did:a", n, &repo);
}
for (rkey, n) in ["1", "2", "3", "4", "5"].iter().enumerate() {
let rkey = format!("3kaaaaaaaaab{rkey}");
state(
&index,
"did:ci",
&rkey,
&uri(n),
json!({"batch": "b".repeat(40)}),
);
}
state(
&index,
"did:ci",
"3kaaaaaaaaac2",
&uri("6"),
json!({"batch": "d".repeat(40)}),
);
// One not allowed to merge saying 6 landed with the five does not
// make its batch five.
let b = json!({"batch": "d".repeat(40)});
state(&index, "did:x", "3kaaaaaaaaac3", &uri("1"), b);
// Pull 1 was turned away for a conflict before it landed.
let conflict = format!("it does not apply on main at 0123: {CONFLICTS} a.rs");
queue("did:ci", "3kaaaaaaaaad1", "1", &conflict);
// Pull 7 was found by halving a batch, and not merged; a record by
// anyone but the runner says nothing.
queue(
"did:ci",
"3kaaaaaaaaad2",
"7",
&format!("its batch failed. {HALVED}"),
);
queue(
"did:x",
"3kaaaaaaaaad3",
"6",
&format!("its batch failed. {HALVED}"),
);
assert_eq!(
ids(&index, "did:a"),
[
("peacemaker", 1),
("good-company", 5),
("smooth-landing", 5),
("bisected", 1),
("first-light", 1)
]
);
}
#[test]
fn badges_count_reviews_and_pulls_from_forks() {
let (_dir, index, repo) = index();
let fork = json!({"name": "r", "key": "f", "createdAt": "2026-01-01T00:00:00Z",
"forkOf": repo});
index.put("did:b", nsid::REPO, "f", &fork).unwrap();
let forked = json!({"uri": "at://did:b/dev.knotless.repo/f", "key": "f"});
pull(&index, "did:b", "1", &forked);
pull(&index, "did:b", "2", &forked);
pull(&index, "did:b", "3", &repo);
let uri = |n: &str| format!("at://did:b/dev.knotless.pull/{n}");
state(&index, "did:a", "3kaaaaaaaaaa2", &uri("1"), json!({}));
state(&index, "did:a", "3kaaaaaaaaaa3", &uri("3"), json!({}));
let comment = |who: &str, rkey: &str, n: &str, at: &str| {
let c = json!({"repo": repo, "subject": uri(n), "body": "lgtm", "createdAt": at});
index.put(who, nsid::COMMENT, rkey, &c).unwrap();
};
// did:c read 1 before it merged and 3 only after; 2 never merged.
comment("did:c", "1", "1", "2026-01-02T10:00:00Z");
comment("did:c", "2", "1", "2026-01-04T10:00:00Z");
comment("did:c", "3", "3", "2026-01-05T00:00:00Z");
comment("did:c", "4", "2", "2026-01-02T10:00:00Z");
assert_eq!(ids(&index, "did:c"), [("second-pair", 1)]);
// did:a's repository had a pull from its fork merged; 3 came from
// its own repository, not a fork.
assert_eq!(
ids(&index, "did:a"),
[("forked-and-fixed", 1), ("first-light", 1)]
);
}
}