johnandersen777.bsky.social/knotless
git that lives in your Atmosphere account
Forked from chadtmiller.com/knotless
0
0
//! This server's own index of the private repositories its members read
//! here: every record in their spaces, synced with the session of the
//! member who last read the repository here ([`App::keep`]), or of its
//! owner, whose own are found as they log in or look at their profile
//! ([`discover`]), so their pages read as a public repository's do, from an
//! index, rather than from every member's repo at each look. A space whose
//! sync fails, as it does once that member is removed, is forgotten, with
//! everything it gave.
//!
//! Nothing written in a space reaches the firehose. Each followed space's
//! authority is asked to tell this server of every write there
//! (`registerNotify`), as is a new private repository's by the command
//! line that makes it, so that it is in the index before anyone looks
//! ([`written`]). Told of one, or every so often in case a notification
//! went missing, the authority is asked which members' repos were written
//! to since the last sync, and each of those is read on through its oplog
//! from the revision it was synced at, checked against the commit its
//! server vouches for ([`knotless::space_sync`]). A notification is only
//! ever a reason to look. One never synced, or whose writes do not check
//! out, is read whole and replaces what it gave before.
use crate::index::Index;
use crate::{App, AppError};
use anyhow::Result;
use atproto::space::LtHash;
use axum::Json;
use axum::extract::State;
use axum::http::{HeaderMap, StatusCode};
use knotless::crypto::PublicKey;
use knotless::env::Ctx;
use knotless::identity;
use knotless::records::nsid;
use knotless::remote::Remote;
use knotless::space::{self, Space};
use knotless::space_sync;
use knotless::url::RepoUrl;
use knotless::xrpc::{Xrpc, XrpcError};
use serde::Deserialize;
use serde_json::{Value, json};
use std::collections::{BTreeMap, HashMap, HashSet};
use std::sync::atomic::Ordering;
use std::sync::{Arc, LazyLock, Mutex};
use std::time::{Duration, Instant};
/// The collections a private repository's pages read, as the appview
/// indexes them from the firehose for public ones; and an organization's
/// records, as its space and its private repositories' spaces keep them,
/// naming its invitations and hidden members.
const COLLECTIONS: [&str; 20] = [
nsid::REPO,
nsid::REDIRECT,
nsid::REF,
nsid::CLAIM,
nsid::CHECK,
nsid::PULL,
nsid::ISSUE,
nsid::COMMENT,
nsid::STATE,
nsid::HIDE,
nsid::BLOCK,
nsid::QUEUE,
nsid::AUTO_MERGE,
nsid::LABEL,
nsid::ASSIGNMENT,
nsid::REACTION,
nsid::ORG,
nsid::ORG_MEMBER,
nsid::ORG_ACCEPTANCE,
// A private repository's site, served to its members.
nsid::SITE,
];
/// How long a space may go unsynced before a member reading it syncs it
/// with their own credential: two of [`follow`]'s sweeps. A space's
/// syncer that fails, as one whose session here lapsed or lacks the
/// scope its space needs does, would otherwise leave it as it was until
/// they log in again, every write since missing from it, to every reader.
pub const STALE: Duration = Duration::from_secs(60);
/// Syncs each space `app` keeps an index of as its authority tells of
/// writes there, and every one of them every `every` besides, for as long
/// as it runs, each as the member it is synced with.
pub async fn follow(app: Arc<App>, every: Duration) {
let Some(index) = app.spaces.clone() else {
return;
};
app.following.store(true, Ordering::Relaxed);
let mut swept: Option<Instant> = None;
loop {
let all = swept.is_none_or(|at| at.elapsed() >= every);
if all {
swept = Some(Instant::now());
}
let due = std::mem::take(&mut *app.due.lock().unwrap());
for (uri, did) in to_sync(app.syncers(), all, &due) {
let ctx = crate::login::syncer_ctx(&app, &did);
// A credential asked for afresh now and then, which the owner's
// PDS gives only to one still a member.
if all {
space::forget(&ctx);
}
let synced = sync(&ctx, &index, &uri).await;
refused_for_scope(&did, &uri, &synced);
let failed = synced.is_err();
if !failed {
app.synced_now(&uri);
}
// Refused, as once the space is deleted or its syncer no longer
// a member, it is followed no more; its authority not reached,
// as while it restarts, it is synced again at the next.
let refused = synced.as_ref().err().is_some_and(space::denied);
let gone = synced.as_ref().err().is_some_and(gone);
// What was written there shows on the next look at it, as does
// what a sync that failed read of its other repos.
if !matches!(synced, Ok(false)) {
app.moved(&uri);
}
crate::telemetry::space_synced(&uri, synced.map(drop));
if gone {
let _ = index.space_gone(&uri);
}
if refused {
app.unfollow(&index, &uri);
}
if !failed {
listen(&app, &ctx, &uri).await;
}
}
let since = swept.map_or(every, |at| at.elapsed());
let left = every.saturating_sub(since);
tokio::select! {
() = tokio::time::sleep(left) => {}
() = app.told.notified() => {}
}
}
}
/// Of `syncers`, each a space and the member it is synced as, those to
/// sync now: every one on a sweep (`all`), else those whose authority told
/// of a write there (`due`). None whose member was refused it for want of
/// scope lately ([`out_of_scope`]): their grant stays as it was until they
/// log in again, and the space's readers sync it meanwhile, once it goes
/// [`STALE`].
fn to_sync(
syncers: Vec<(String, String)>,
all: bool,
due: &HashSet<String>,
) -> Vec<(String, String)> {
let now = |(uri, _): &(String, String)| all || due.contains(uri);
let asked = |(uri, did): &(String, String)| !out_of_scope(did, uri);
syncers.into_iter().filter(now).filter(asked).collect()
}
/// Of [`space::denied`], what says the space itself is no more, rather
/// than that its syncer may no longer read it.
fn gone(err: &anyhow::Error) -> bool {
let said = XrpcError::of(err).map(|e| e.error.as_str());
matches!(said, Some("SpaceNotFound" | "SpaceDeleted"))
}
/// How long this server's asking a space's authority to tell it of writes
/// there is taken to hold before it asks again: the reference PDS keeps
/// one for a day.
pub const RENEW: Duration = Duration::from_secs(12 * 60 * 60);
/// Asks the authority of the space `uri`, as its syncer (`ctx`), to tell
/// this server of each write there, unless it did within [`RENEW`].
async fn listen(app: &App, ctx: &Ctx, uri: &str) {
let mut registered = app.registered.lock().await;
let asked = registered.get(uri).is_some_and(|at| at.elapsed() < RENEW);
if asked {
return;
}
let did = app.gateway.as_ref().map(|gw| gw.did.as_str());
let service = format!("{}{}", did.unwrap_or_default(), space::APPVIEW);
let heard = register(ctx, uri, &service).await;
crate::telemetry::space_unheard(uri, heard);
// Asked again after as long whether or not it was heard, so that an
// authority without notifications is not asked at every sync.
registered.insert(uri.to_string(), Instant::now());
}
/// [`listen`], whatever came before.
async fn register(ctx: &Ctx, uri: &str, service: &str) -> Result<String> {
let joined = space::joined(ctx, uri).await?;
let owner = identity::resolve(ctx, authority(uri)).await?;
let host = Xrpc::new(ctx.http.clone(), &owner.pds);
joined.register_notify(&host, &owner.did, service).await
}
/// The account a space is under, its authority.
fn authority(uri: &str) -> &str {
let authority = uri.trim_start_matches("at://").split('/').next();
authority.unwrap_or_default()
}
#[derive(Deserialize)]
pub struct Told {
space: String,
}
/// `com.atproto.space.notifyWrite`, from a space's authority: a member's
/// repo there moved. A space followed is synced now; one not yet, as a
/// private repository just made is, is followed as its owner, and synced
/// now, if they are logged in here. Anyone else's is no one's to read here.
pub async fn written(
State(app): State<Arc<App>>,
headers: HeaderMap,
Json(told): Json<Told>,
) -> Result<Json<Value>, AppError> {
let (index, owner) = from_authority(&app, &headers, &told.space, NOTIFY_WRITE).await?;
app.live.written(&told.space);
let followed = app.syncers().iter().any(|(uri, _)| *uri == told.space);
let ctx = crate::login::ctx_for(&app, &owner);
let theirs = knotless::session::load(&ctx, &owner).is_some();
if !followed && theirs {
let follows = index.follow_space(&told.space, &owner);
follows
.map_err(anyhow::Error::from)
.map_err(AppError::internal)?;
}
app.due.lock().unwrap().insert(told.space);
app.told.notify_one();
Ok(Json(json!({})))
}
/// `com.atproto.space.notifySpaceDeleted`, from a space's authority: what
/// it gave goes, and it is followed no more.
pub async fn deleted(
State(app): State<Arc<App>>,
headers: HeaderMap,
Json(told): Json<Told>,
) -> Result<Json<Value>, AppError> {
let (index, _) = from_authority(&app, &headers, &told.space, NOTIFY_DELETED).await?;
app.unfollow(&index, &told.space);
let gone = index.space_gone(&told.space);
gone.map_err(anyhow::Error::from)
.map_err(AppError::internal)?;
app.registered.lock().await.remove(&told.space);
Ok(Json(json!({})))
}
pub const NOTIFY_WRITE: &str = "com.atproto.space.notifyWrite";
pub const NOTIFY_DELETED: &str = "com.atproto.space.notifySpaceDeleted";
/// This server's index of spaces, and the account that signed the
/// request's token for `lxm`, to this server's [`space::APPVIEW`]: the
/// authority of `space`, a private repository's or its organization's, or
/// it is refused.
async fn from_authority(
app: &App,
headers: &HeaderMap,
space: &str,
lxm: &str,
) -> Result<(Arc<Index>, String), AppError> {
let none = AppError(StatusCode::NOT_FOUND, "this server syncs no spaces".into());
let index = app.spaces.clone().ok_or(none)?;
let did = app.gateway.as_ref().map(|gw| gw.did.as_str());
let aud = format!("{}{}", did.unwrap_or_default(), space::APPVIEW);
let caller = crate::gateway::caller_for(app, headers, &aud, lxm).await?;
let theirs = space.starts_with(&space::uri(&caller, ""));
if !theirs && space != knotless::org::space_uri(&caller) {
let why = "only a private repository's authority tells of its space";
return Err(AppError(StatusCode::FORBIDDEN, why.into()));
}
Ok((index, caller))
}
/// How often an owner's spaces are listed again, as they log in or look at
/// their profile, for private repositories made since.
pub const DISCOVER: Duration = Duration::from_secs(60);
/// Follows, and syncs, each space `did` owns that this server does not yet,
/// as them: their private repositories, so that their profile lists them
/// from the index, and keeps whether their PDS has spaces at all, for the
/// pages to say whether they can make one ([`App::has_spaces`]). Of the
/// spaces followed that are someone else's, each they may be a member of
/// ([`Index::maybe_member`]) is asked of its authority as them, so the
/// repositories shared with them are listed to them before they open one,
/// and no longer once they are refused. In the background, at most once
/// every [`DISCOVER`], so whoever looks never waits on it, and one at a
/// time for each account, so that an answer asked before a removal is
/// never kept after one asked since.
pub fn discover(app: &Arc<App>, did: &str) {
let Some(index) = app.spaces.clone().filter(|_| app.discovering(did)) else {
return;
};
let app = app.clone();
let did = did.to_string();
tokio::spawn(async move {
let found = owned(&app, &index, &did).await;
crate::telemetry::space_synced(&did, found);
});
}
/// [`discover`], waited on.
async fn owned(app: &App, index: &Index, did: &str) -> Result<()> {
let lock = app.discover_lock(did);
let _held = lock.lock().await;
let ctx = crate::login::ctx_for(app, did);
let me = knotless::session::me(&ctx).await?;
let followed: HashSet<String> = index.spaces()?.into_iter().map(|(uri, _)| uri).collect();
let owned = Remote::own_spaces(&ctx, &me).await?;
app.found_spaces(did, owned.is_some());
// One its authority said is deleted, which a PDS can go on naming, is
// not followed again only to fail.
let gone = index.gone_spaces()?;
for uri in owned
.into_iter()
.flatten()
.filter(|uri| !gone.contains(uri))
{
index.follow_space(&uri, did)?;
let new = !followed.contains(&uri);
let synced = if new {
sync(&ctx, index, &uri).await.map(drop)
} else {
Ok(())
};
// One that fails keeps none of the rest from being looked at; its
// sync failing, [`follow`] forgets it.
crate::telemetry::space_synced(&uri, synced);
}
// Asked afresh, as a reading is, since a kept credential outlives a
// removal.
space::forget(&ctx);
let members = index.maybe_member(did)?.into_iter();
for uri in members.filter(|uri| !out_of_scope(did, uri)) {
let joined = space::joined(&ctx, &uri).await;
refused_for_scope(did, &uri, &joined);
// Anything but an answer leaves it as it was, to ask again.
if joined.is_ok() {
index.read_by(&uri, did)?;
// Followed by no one, as after the index is rebuilt, it is
// synced as them, so what they read is listed again.
if !followed.contains(&uri) {
index.follow_space(&uri, did)?;
crate::telemetry::space_synced(&uri, sync(&ctx, index, &uri).await.map(drop));
}
}
// Refused by its authority, they read it no more; its authority not
// reached, or the session not let ask, says nothing either way.
crate::telemetry::reader_checked(&uri, did, &joined);
if joined.as_ref().err().is_some_and(space::denied) {
index.unread_by(&uri, did)?;
}
}
Ok(())
}
/// The spaces each account's session was refused for want of scope, by
/// account and space, and when: not asked again until [`SCOPE_ASKED`]
/// passes, or it logs in again ([`in_scope_again`]), as its grant stays as
/// it was until then, and each asking costs a call to its PDS, every
/// [`DISCOVER`] as a reader, every sweep of [`follow`] as a syncer.
static OUT_OF_SCOPE: LazyLock<Mutex<HashMap<(String, String), Instant>>> =
LazyLock::new(Mutex::default);
/// How long a space a session was refused for want of scope is not asked.
const SCOPE_ASKED: Duration = Duration::from_secs(60 * 60);
/// Whether `did`'s session was refused the space `uri` for want of scope
/// within [`SCOPE_ASKED`].
fn out_of_scope(did: &str, uri: &str) -> bool {
let refused = OUT_OF_SCOPE.lock().unwrap();
let at = refused.get(&(did.to_string(), uri.to_string()));
at.is_some_and(|at| at.elapsed() < SCOPE_ASKED)
}
/// Keeps that `did`'s session was refused the space `uri` for want of
/// scope, if `joined` says it was.
fn refused_for_scope<T>(did: &str, uri: &str, joined: &Result<T>) {
if joined
.as_ref()
.err()
.is_some_and(knotless::oauth::out_of_scope)
{
let key = (did.to_string(), uri.to_string());
OUT_OF_SCOPE.lock().unwrap().insert(key, Instant::now());
}
}
/// Forgets the spaces `did`'s session was refused for want of scope: it
/// was granted anew, as by logging in again, so each is asked again.
pub fn in_scope_again(did: &str) {
OUT_OF_SCOPE.lock().unwrap().retain(|(of, _), _| of != did);
}
/// Takes in what was written to the space `uri` since it was last synced,
/// and says whether anything was.
pub async fn sync(ctx: &Ctx, index: &Index, uri: &str) -> Result<bool> {
let joined = space::joined(ctx, uri).await?;
sync_with(ctx, index, &joined).await
}
/// [`sync`], with `joined`, the space as some member reads it. One sync
/// of a space at a time, so that none puts back what it read over what a
/// later one did. A repo that fails keeps none of the rest from being
/// read: it is the error, and the space's cursor stays where it was, so
/// it is listed, and read, again at the next.
pub async fn sync_with(ctx: &Ctx, index: &Index, joined: &Space) -> Result<bool> {
let uri = joined.uri.as_str();
let lock = index.syncing(uri);
let _held = lock.lock().await;
let owner = identity::resolve(ctx, authority(uri)).await?;
let host = Xrpc::new(ctx.http.clone(), &owner.pds);
let since = index.spaces()?.into_iter().find(|(u, _)| u == uri);
let since = since.and_then(|(_, cursor)| cursor);
let (writers, now) = joined
.writers_since(&host, &owner.did, since.as_deref())
.await?;
// A repo written to again while the listing was read is listed twice:
// the later says where it is now. One read to there already has
// nothing new.
let writers: BTreeMap<String, String> =
writers.into_iter().map(|w| (w.did, w.repo_rev)).collect();
let fresh = |(did, rev): &(String, String)| {
let kept = index.space_writer(uri, did).ok().flatten();
kept.is_none_or(|(at, _)| at != *rev)
};
let writers: Vec<_> = writers.into_iter().filter(fresh).collect();
let wrote = !writers.is_empty();
let mut failed = None;
for (did, _) in writers {
let read = member(ctx, index, joined, &did).await;
failed = failed.or(read.err());
}
failed.map_or(Ok(()), Err)?;
if let Some(now) = now {
index.set_space_cursor(uri, &now)?;
}
Ok(wrote)
}
/// Takes in `did`'s repo in `joined`: read on from where it was last
/// synced to, or whole, or, should it not come to its commit whole, from
/// the start of its oplog, where each record's CID is vouched for apart
/// from its bytes.
async fn member(ctx: &Ctx, index: &Index, joined: &Space, did: &str) -> Result<()> {
let theirs = remote(ctx, joined, did).await?;
let kem = atproto::did::kem_key(&identity::document(ctx, did).await?)?;
let uri = &joined.uri;
let read = match index.space_writer(uri, did)? {
Some((rev, state)) => {
let read = read_on(index, uri, &theirs, &kem, &rev, &state).await;
crate::telemetry::read_whole(uri, did, &read);
read.is_ok()
}
None => false,
};
if !read {
let read = whole(index, uri, &theirs, &kem).await;
crate::telemetry::read_from_start(uri, did, &read);
if read.is_err() {
from_start(index, uri, &theirs, &kem).await?;
}
}
Ok(())
}
/// `did`'s repo in the space `joined`, read with its credential.
pub async fn remote(ctx: &Ctx, joined: &Space, did: &str) -> Result<Remote> {
let id = identity::resolve(ctx, did).await?;
Ok(Remote {
xrpc: Xrpc::new(ctx.http.clone(), &id.pds),
identity: id,
url: RepoUrl {
authority: did.to_string(),
name: String::new(),
},
space: Some(joined.at(did)),
})
}
/// Whether the pages read records of `collection`.
fn indexed(collection: &str) -> bool {
COLLECTIONS.contains(&collection)
}
/// Reads on from `rev` in `theirs`'s oplog, where its set hash's state was
/// `state`: each write since, checked against the commit they come to,
/// and each record they leave, against the CID its last write gave it.
async fn read_on(
index: &Index,
uri: &str,
theirs: &Remote,
kem: &PublicKey,
rev: &str,
state: &[u8],
) -> Result<()> {
let hash = LtHash::from_state(state)?;
let read = space_sync::read_on(theirs, kem, Some(rev), hash).await?;
let did = theirs.did();
let (changes, unreadable) = read.changes(indexed)?;
crate::telemetry::unreadable(uri, did, &unreadable);
for change in changes {
let (collection, rkey) = (&change.collection, &change.rkey);
match &change.value {
Some(value) => index.put_from_space(uri, did, collection, rkey, value)?,
None => index.delete_from_space(uri, did, collection, rkey)?,
}
}
index.set_space_writer(uri, did, &read.commit.rev, &read.hash.state())?;
Ok(())
}
/// Reads `theirs` whole, in place of all it gave before: what it no longer
/// has is taken out.
async fn whole(index: &Index, uri: &str, theirs: &Remote, kem: &PublicKey) -> Result<()> {
let whole = space_sync::whole(theirs, kem).await?;
let did = theirs.did();
crate::telemetry::unreadable(uri, did, &whole.unreadable);
let records = whole.records.iter().filter(|r| indexed(&r.collection));
let records = records.map(|r| (&r.collection, &r.rkey, &r.value));
let commit = (whole.commit.rev.as_str(), &whole.hash);
replace(index, uri, did, records, commit)
}
/// Reads `theirs` from the start of its oplog, in place of all it gave
/// before, as [`whole`] does: for a repo whose archive does not come to
/// its commit, as one holding a record its server keeps cut short, whose
/// bytes no longer come to the CID it was written with. The oplog gives
/// each record's CID apart from its bytes, which the commit vouches for
/// all together, and each record's value is checked against its own:
/// one that does not come to it is left out. It comes to the commit only
/// if the oplog still holds every write the repo was made of.
async fn from_start(index: &Index, uri: &str, theirs: &Remote, kem: &PublicKey) -> Result<()> {
let read = space_sync::read_on(theirs, kem, None, LtHash::default()).await?;
let did = theirs.did();
let (changes, unreadable) = read.changes(indexed)?;
crate::telemetry::unreadable(uri, did, &unreadable);
let records = changes.iter().filter_map(|c| {
let value = c.value.as_ref()?;
Some((&c.collection, &c.rkey, value))
});
replace(
index,
uri,
did,
records,
(read.commit.rev.as_str(), &read.hash),
)
}
/// Puts `records` in the index as all `did`'s repo in the space `uri`
/// holds, at the revision and set hash its commit came to: what it gave
/// before and no longer has is taken out.
fn replace<'a>(
index: &Index,
uri: &str,
did: &str,
records: impl Iterator<Item = (&'a String, &'a String, &'a Value)>,
(rev, hash): (&str, &LtHash),
) -> Result<()> {
let mut gone: HashSet<(String, String)> = index.space_records(uri, did)?.into_iter().collect();
for (collection, rkey, value) in records {
index.put_from_space(uri, did, collection, rkey, value)?;
gone.remove(&(collection.clone(), rkey.clone()));
}
for (collection, rkey) in gone {
index.delete_from_space(uri, did, &collection, &rkey)?;
}
index.set_space_writer(uri, did, rev, &hash.state())?;
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn a_space_out_of_scope_is_not_asked_again_a_while() {
let (did, uri) = ("did:plc:s", "at://did:plc:o/space/dev.knotless.org/self");
refused_for_scope::<()>(did, uri, &Ok(()));
refused_for_scope::<()>(did, uri, &Err(anyhow::anyhow!("connection refused")));
assert!(!out_of_scope(did, uri));
let scope = anyhow::anyhow!("InvalidToken (403): Scope does not permit reading {uri}");
refused_for_scope::<()>(did, uri, &Err(scope));
assert!(out_of_scope(did, uri));
assert!(!out_of_scope("did:plc:t", uri));
let key = (did.to_string(), uri.to_string());
let long_ago = Instant::now() - SCOPE_ASKED;
OUT_OF_SCOPE.lock().unwrap().insert(key, long_ago);
assert!(!out_of_scope(did, uri));
}
#[test]
fn syncers_out_of_scope_wait_till_they_log_in_again() {
let did = "did:plc:syncer";
let (uri, other) = (
"at://did:plc:o/space/dev.knotless.space/r",
"at://did:plc:o/space/dev.knotless.space/s",
);
let syncers = || vec![(uri.into(), did.into()), (other.into(), did.into())];
let uris = |s: Vec<(String, String)>| s.into_iter().map(|(u, _)| u).collect::<Vec<_>>();
let none = HashSet::new();
assert_eq!(uris(to_sync(syncers(), true, &none)), [uri, other]);
let due = HashSet::from([other.to_string()]);
assert_eq!(uris(to_sync(syncers(), false, &due)), [other]);
let scope = anyhow::anyhow!("InvalidToken (403): Scope does not permit reading {uri}");
refused_for_scope::<()>(did, uri, &Err(scope));
assert_eq!(uris(to_sync(syncers(), true, &none)), [other]);
in_scope_again(did);
assert_eq!(uris(to_sync(syncers(), true, &none)), [uri, other]);
}
}