johnandersen777.bsky.social/knotless

Switch repository

Loading…

//! Asking after an account said to be deleted, against a stand-in PLC
//! directory that is also each account's host: only an answer that says
//! so is gone, and one that cannot be had is asked again, then kept.

use axum::extract::{Query, State};
use axum::http::{StatusCode, Uri};
use axum::response::{IntoResponse, Response};
use axum::{Json, Router};
use knotless_serve::App;
use knotless_serve::accounts::{self, Asked, Patience, Settled};
use serde_json::{Value, json};
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;

/// Each account by what its identity and its host say of it.
const HERE: &str = "did:plc:hereaaaaaaaaaaaaaaaaaaaa";
const DELETED: &str = "did:plc:deletedaaaaaaaaaaaaaaaa";
const DEACTIVATED: &str = "did:plc:deactivatedaaaaaaaaaaaa";
const SUSPENDED: &str = "did:plc:suspendedaaaaaaaaaaaaaa";
const TAKENDOWN: &str = "did:plc:takendownaaaaaaaaaaaaaa";
const NO_REPO: &str = "did:plc:norepoaaaaaaaaaaaaaaaaa";
const TOMBSTONE: &str = "did:plc:tombstoneaaaaaaaaaaaaaa";
const NEVER: &str = "did:plc:neveraaaaaaaaaaaaaaaaaa";
const PLC_DOWN: &str = "did:plc:plcdownaaaaaaaaaaaaaaaa";
const HOST_DOWN: &str = "did:plc:hostdownaaaaaaaaaaaaaaa";
const UNREACHABLE: &str = "did:plc:unreachableaaaaaaaaaaaa";
const ODD: &str = "did:plc:oddaaaaaaaaaaaaaaaaaaaaa";
const NO_HOST: &str = "did:plc:nohostaaaaaaaaaaaaaaaaa";
/// Its PLC fails the first time it is asked, then its host says it was
/// deleted.
const FLAKY: &str = "did:plc:flakyaaaaaaaaaaaaaaaaaa";

struct Stand {
    base: String,
    /// A port nothing listens on.
    closed: String,
    asked: AtomicUsize,
}

fn doc(did: &str, pds: &str) -> Value {
    json!({
        "id": did,
        "service": [{
            "id": "#atproto_pds",
            "type": "AtprotoPersonalDataServer",
            "serviceEndpoint": pds,
        }],
    })
}

/// A status the host holds `did` at, not active.
fn inactive(did: &str, status: &str) -> Response {
    Json(json!({ "did": did, "active": false, "status": status })).into_response()
}

async fn answer(
    State(stand): State<Arc<Stand>>,
    uri: Uri,
    Query(q): Query<HashMap<String, String>>,
) -> Response {
    let refused = |status: StatusCode, error: &str| {
        let body = json!({ "error": error, "message": error });
        (status, Json(body)).into_response()
    };
    if uri.path() == "/xrpc/com.atproto.sync.getRepoStatus" {
        let did = q["did"].as_str();
        return match did {
            HERE => Json(json!({ "did": did, "active": true })).into_response(),
            DELETED | FLAKY => inactive(did, "deleted"),
            DEACTIVATED => inactive(did, "deactivated"),
            SUSPENDED => inactive(did, "suspended"),
            TAKENDOWN => inactive(did, "takendown"),
            NO_REPO => refused(StatusCode::BAD_REQUEST, "RepoNotFound"),
            HOST_DOWN => refused(StatusCode::BAD_GATEWAY, "InternalServerError"),
            _ => Json(json!({ "did": did })).into_response(),
        };
    }
    let did = uri.path().trim_start_matches('/');
    match did {
        TOMBSTONE => refused(StatusCode::GONE, "DID not available"),
        PLC_DOWN => refused(StatusCode::SERVICE_UNAVAILABLE, "down"),
        FLAKY if stand.asked.fetch_add(1, Ordering::SeqCst) == 0 => {
            refused(StatusCode::INTERNAL_SERVER_ERROR, "busy")
        }
        UNREACHABLE => Json(doc(did, &stand.closed)).into_response(),
        NO_HOST => Json(json!({ "id": did, "service": [] })).into_response(),
        NEVER => refused(StatusCode::NOT_FOUND, "DID not registered"),
        _ => Json(doc(did, &stand.base)).into_response(),
    }
}

/// The stand-in, served on `rt`, and an appview that reads identities
/// from it, its cache under `dir`.
fn stand_in(rt: &tokio::runtime::Runtime, dir: &std::path::Path) -> Arc<App> {
    let listener = rt
        .block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
        .unwrap();
    let base = format!("http://{}", listener.local_addr().unwrap());
    let closed = {
        let spare = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        format!("http://{}", spare.local_addr().unwrap())
    };
    let stand = Arc::new(Stand {
        base: base.clone(),
        closed,
        asked: AtomicUsize::new(0),
    });
    let router = Router::new().fallback(answer).with_state(stand);
    rt.spawn(async move { axum::serve(listener, router).await });
    let (config, cache) = (dir.join("config"), dir.join("cache"));
    let env = knotless::env::Env::from_vars(|name| match name {
        "KNOTLESS_PLC_URL" => Some(base.clone()),
        "KNOTLESS_CONFIG_DIR" => Some(config.to_string_lossy().into()),
        "KNOTLESS_CACHE_DIR" => Some(cache.to_string_lossy().into()),
        "KNOTLESS_APPVIEW" => Some(String::new()),
        _ => None,
    });
    let ctx = knotless::env::Ctx::new(env);
    App::build(ctx, Duration::ZERO, None, None)
}

/// Quick to try again, and soon done.
const BRIEF: Patience = Patience {
    first: Duration::from_millis(10),
    most: Duration::from_millis(20),
    tries: 3,
};

#[test]
fn only_an_answer_that_says_so_is_gone() {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let dir = tempfile::tempdir().unwrap();
    let app = stand_in(&rt, dir.path());
    let asked = |did: &str| rt.block_on(accounts::gone(&app.ctx, did));
    // Active, or inactive in a way that can be undone: still there.
    for did in [HERE, DEACTIVATED, SUSPENDED, TAKENDOWN] {
        assert_eq!(asked(did), Asked::Active, "{did}");
    }
    for did in [DELETED, NO_REPO, TOMBSTONE, NEVER] {
        assert_eq!(asked(did), Asked::Gone, "{did}");
    }
    for did in [PLC_DOWN, HOST_DOWN, UNREACHABLE, ODD, NO_HOST] {
        let asked = asked(did);
        assert!(matches!(asked, Asked::Unknown(_)), "{did}: {asked:?}");
    }
}

#[test]
fn an_account_not_settled_is_asked_again_then_kept() {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let dir = tempfile::tempdir().unwrap();
    let app = stand_in(&rt, dir.path());
    // A mirror of each, which goes only with an account gone.
    let mirrors = |did: &str| {
        let dir = did.replace(':', "_");
        app.ctx.env.cache_dir.join("mirrors").join(dir)
    };
    for did in [HERE, DEACTIVATED, FLAKY, PLC_DOWN] {
        std::fs::create_dir_all(mirrors(did).join("repo.git")).unwrap();
    }
    let settle = |did: &str| rt.block_on(accounts::settle(&app, did, BRIEF));
    assert_eq!(settle(HERE), Settled::Kept);
    // Deactivated where it lives: kept, with its mirrors.
    assert_eq!(settle(DEACTIVATED), Settled::Kept);
    // PLC fails once, then the host says it was deleted.
    assert_eq!(settle(FLAKY), Settled::Purged);
    // Never answered: asked each time, then kept.
    assert_eq!(settle(PLC_DOWN), Settled::GaveUp);
    let kept = [HERE, DEACTIVATED, FLAKY, PLC_DOWN].map(|did| mirrors(did).exists());
    assert_eq!(kept, [true, true, false, true]);
}

#[test]
fn patience_is_about_half_an_hour() {
    let p = Patience::default();
    assert_eq!((p.first.as_secs(), p.most.as_secs(), p.tries), (30, 600, 6));
}