johnandersen777.bsky.social/knotless
git that lives in your Atmosphere account
Forked from chadtmiller.com/knotless
0
0
//! What an account's host says of it, from the `#account` events the feeds
//! carry. One not active there, deactivated, taken down or suspended, which
//! it may be undone, is left out of every read until it is active again,
//! with what was indexed of it kept. One deleted is gone for good: what
//! this server indexed from it, the mirrors of its repositories and the
//! sessions it kept for it go with it.
//!
//! An event saying an account was deleted is not taken at its word: the
//! host its identity names now is asked first ([`gone`]), and only an
//! answer that says so drops anything. One that cannot be had, PLC or the
//! host down or answering oddly, is asked for again, then the account kept
//! ([`settle`]), since nothing dropped of it would come back.
use crate::App;
use crate::telemetry;
use knotless::env::Ctx;
use knotless::telemetry::Level;
use knotless::xrpc::{Xrpc, XrpcError};
use serde_json::Value;
use std::path::Path;
use std::sync::Arc;
use std::time::Duration;
/// Takes in that `did`'s host says `status` of it.
pub(crate) fn changed(app: &Arc<App>, did: &str, status: &str) {
if status != "deleted" {
for index in [&app.index, &app.spaces].into_iter().flatten() {
let _ = index.set_status(did, status);
}
return;
}
let (app, did) = (app.clone(), did.to_string());
tokio::spawn(async move { settle(&app, &did, Patience::default()).await });
}
/// What the host `did`'s identity names now says of it.
#[derive(Debug, Clone, PartialEq)]
pub enum Asked {
/// Gone: the host has no such repo (`RepoNotFound`) or says it was
/// deleted (`active: false`, `status: deleted`), or the identity itself
/// is gone (a PLC tombstone, `410`, or nothing there, `404`).
Gone,
/// The host has it: active, as it is where it moved to, or inactive in
/// a way that can be undone (deactivated, suspended, taken down).
Active,
/// Nothing that says either: the identity or the host could not be
/// read, or answered with what says neither. Why, for the log.
Unknown(String),
}
/// Asks after `did`, said to be deleted, where its identity says it lives
/// now ([`Asked`]): one that moved was deleted only where it was before.
pub async fn gone(ctx: &Ctx, did: &str) -> Asked {
let doc = match knotless::identity::document(ctx, did).await {
Ok(doc) => doc,
Err(e) if matches!(http_status(&e), Some(404 | 410)) => return Asked::Gone,
Err(e) => return Asked::Unknown(format!("{e:#}")),
};
let Some(pds) = atproto::did::pds(&doc) else {
return Asked::Unknown(format!("{did} names no #atproto_pds service"));
};
let xrpc = Xrpc::new(ctx.http.clone(), &pds);
let asked = [("did", did)];
let status = xrpc.query::<Value>("com.atproto.sync.getRepoStatus", &asked, None);
match status.await {
Ok(status) => match (status["active"].as_bool(), status["status"].as_str()) {
(Some(true), _) => Asked::Active,
(Some(false), Some("deleted")) => Asked::Gone,
// Deactivated, suspended or taken down, which can be undone:
// the account is still there.
(Some(false), why) => {
let said = "an account inactive at its host, not deleted";
telemetry::account_asked(Level::INFO, said, did, why, None, None);
Asked::Active
}
(None, _) => Asked::Unknown(format!("{pds} answered {status}")),
},
Err(e) => match XrpcError::of(&e) {
Some(refused) if refused.error == "RepoNotFound" => Asked::Gone,
_ => Asked::Unknown(format!("{e:#}")),
},
}
}
/// The HTTP status a request in `e` was answered with, if it was.
fn http_status(e: &anyhow::Error) -> Option<u16> {
let answered = e.chain().find_map(|c| c.downcast_ref::<reqwest::Error>());
answered
.and_then(reqwest::Error::status)
.map(|s| s.as_u16())
}
/// How long to keep asking after an account no answer has settled.
#[derive(Debug, Clone, Copy)]
pub struct Patience {
/// The wait before asking again, doubled each time.
pub first: Duration,
/// The longest wait.
pub most: Duration,
/// How many times to ask in all.
pub tries: u32,
}
impl Default for Patience {
/// Six asks over about half an hour, the last waits capped at ten
/// minutes: long enough for PLC or a host to come back.
fn default() -> Patience {
Patience {
first: Duration::from_secs(30),
most: Duration::from_secs(600),
tries: 6,
}
}
}
/// What came of an account said to be deleted.
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum Settled {
/// Its host has it: it stays.
Kept,
/// Gone: everything held from it was dropped ([`purge`]).
Purged,
/// No answer said, however often asked: it stays, as nothing dropped
/// of it would come back.
GaveUp,
}
/// Asks after `did` until an answer says, waiting longer each time, and
/// drops what is held from it only if it is [`Asked::Gone`].
pub async fn settle(app: &App, did: &str, patience: Patience) -> Settled {
let mut wait = patience.first;
for tried in 1..=patience.tries {
match gone(&app.ctx, did).await {
Asked::Active => {
let said = "an account said to be deleted is there at its host";
telemetry::account_asked(Level::INFO, said, did, None, Some(tried), None);
return Settled::Kept;
}
Asked::Gone => {
purge(app, did);
let said = "an account was deleted";
telemetry::account_asked(Level::INFO, said, did, None, Some(tried), None);
return Settled::Purged;
}
Asked::Unknown(why) if tried < patience.tries => {
let said = "an account said to be deleted, asked after again later";
telemetry::account_asked(Level::WARN, said, did, None, Some(tried), Some(why));
tokio::time::sleep(wait).await;
wait = (wait * 2).min(patience.most);
}
Asked::Unknown(why) => {
let said = "an account said to be deleted, never settled: kept";
telemetry::account_asked(Level::ERROR, said, did, None, Some(tried), Some(why));
}
}
}
Settled::GaveUp
}
/// Drops everything this server holds from `did`, deleted: what both
/// indexes have of it, the mirrors of its repositories, whoever's they were
/// read for, and its sessions, those of the pages, of MCP and of the
/// gateway it delegated to.
pub(crate) fn purge(app: &App, did: &str) {
for index in [&app.index, &app.spaces].into_iter().flatten() {
let _ = index.purge(did);
}
// Each closure on its call's line, as one may find nothing to call it on.
let mut repos = app.repos.lock().unwrap();
repos.retain(|(owner, _), _| owner != did);
drop(repos);
let mut viewed = app.viewed.lock().unwrap();
viewed.retain(|(_, owner, _), _| owner != did);
drop(viewed);
app.discovered.lock().unwrap().remove(did);
crate::private::forget(app, did);
let env = &app.ctx.env;
let dir = did.replace(':', "_");
// The pages' session, and MCP's, each in a directory of its own.
for kept in [&env.config_dir, &env.cache_dir] {
let _ = std::fs::remove_dir_all(kept.join("web").join(&dir));
let _ = std::fs::remove_dir_all(kept.join(MCP).join(&dir));
}
knotless::session::remove(&app.ctx, did);
// Its mirrors: this server's, and those read as each account here.
unmirror(&env.cache_dir, &dir);
for under in readers(&env.cache_dir) {
let each = std::fs::read_dir(under).into_iter().flatten().flatten();
for reader in each {
unmirror(&reader.path(), &dir);
}
}
}
/// Where MCP over HTTP keeps what it does for each account, under the
/// server's config and cache directories (`knotless_mcp::http`).
const MCP: &str = "mcp";
/// Where the accounts this server reads as keep their caches, each in a
/// directory of its own: the server's, the pages' and MCP's.
fn readers(cache: &Path) -> [std::path::PathBuf; 3] {
[cache.to_path_buf(), cache.join("web"), cache.join(MCP)]
}
/// Forgets `did`, whose owner deleted its knotless data: logs it out of
/// the pages and of MCP, revoking those sessions at its PDS, then drops
/// all else this server holds from it, as for an account deleted
/// ([`purge`]), and refuses the pages' cookies signed for it before.
pub(crate) async fn forget(app: &App, did: &str) {
let pages = crate::login::ctx_for(app, did);
let mut mcp = pages.clone();
let dir = did.replace(':', "_");
mcp.env.config_dir = app.ctx.env.config_dir.join(MCP).join(&dir);
for ctx in [&pages, &mcp] {
let _ = knotless::session::logout(ctx, did).await;
}
purge(app, did);
crate::login::refuse_cookies(app, did);
}
/// Removes the mirrors of the account kept as `dir` under `cache`.
fn unmirror(cache: &Path, dir: &str) {
let _ = std::fs::remove_dir_all(cache.join("mirrors").join(dir));
}
/// [`repo_deleted`], of each of `keys`.
pub(crate) fn repos_deleted(app: &Arc<App>, did: &str, keys: &[String]) {
for key in keys {
repo_deleted(app, did, key);
}
}
/// Takes in that `did` deleted the record of its repository `key`: once
/// its host holds no repository record with that key, as one renamed
/// does, what this server holds of the repository goes ([`forget_repo`]).
pub(crate) fn repo_deleted(app: &Arc<App>, did: &str, key: &str) {
let (app, did, key) = (app.clone(), did.to_string(), key.to_string());
tokio::spawn(async move {
if repo_gone(&app.ctx, &did, &key).await {
forget_repo(&app, &did, &key);
}
});
}
/// Whether `did`'s host lists no repository record with `key`: not when
/// it cannot be asked.
async fn repo_gone(ctx: &Ctx, did: &str, key: &str) -> bool {
let url = knotless::url::RepoUrl {
authority: did.to_string(),
name: String::new(),
};
let listed = async {
let remote = knotless::remote::Remote::connect(ctx, &url).await?;
remote.list_repos().await
};
let listed = listed.await;
listed.is_ok_and(|repos| repos.iter().all(|r| r.value.key != key))
}
/// Drops what this server holds of `did`'s repository `key`, deleted:
/// what the index has of it ([`crate::index::Index::forget_repo`]), and
/// its mirrors, this server's and those read as each account here.
pub(crate) fn forget_repo(app: &App, did: &str, key: &str) {
let forgot = app.index.as_ref().map(|index| index.forget_repo(did, key));
// Moved to another account, or under another name again: kept.
if !matches!(forgot, Some(Ok(true))) {
return;
}
app.repos
.lock()
.unwrap()
.retain(|(owner, _), _| owner != did);
let env = &app.ctx.env;
let dir = did.replace(':', "_");
let mirror = |cache: &Path| cache.join("mirrors").join(&dir).join(format!("{key}.git"));
let _ = std::fs::remove_dir_all(mirror(&env.cache_dir));
for under in readers(&env.cache_dir) {
let each = std::fs::read_dir(under).into_iter().flatten().flatten();
for reader in each {
let _ = std::fs::remove_dir_all(mirror(&reader.path()));
}
}
}