ipx/crates/directory/src/lib.rs

//! ipx's directory service (#204): a catalogue of public podcasts and blogs that ipx servers
//! share. It holds metadata only, never a file: each server still fetches its own feeds. ipx
//! servers send which public feeds their people subscribe to, the directory fetches each
//! feed's title, artwork, description and latest items on a slow schedule, and any server
//! holding a token can browse it.
//!
//! The types here are the wire format, shared with ipx, which asks the directory on its
//! people's behalf.

pub mod apple;
pub mod config;
pub mod metrics;
pub mod otel;
mod site;
pub mod store;

use std::sync::{Arc, RwLock};
use std::time::{Duration, Instant};

use axum::extract::{Path, Query, Request, State};
use axum::http::{StatusCode, header};
use axum::middleware::Next;
use axum::response::{IntoResponse, Response};
use axum::routing::{delete, get, post};
use axum::{Extension, Json, Router};
use serde::{Deserialize, Serialize};

use config::{Access, Token};
use store::{Refreshed, Store};

/// A feed as the directory lists it.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Listing {
    /// The first 64 bits of the SHA-256 of its normalised address (`store::feed_id`).
    pub id: String,
    pub url: String,
    pub title: Option<String>,
    pub image: Option<String>,
    /// Apple's category and the subcategory within it, from the feed's own iTunes category.
    pub category: Option<String>,
    pub subcategory: Option<String>,
    /// Any audio or video in its latest items.
    pub podcast: bool,
    /// Summed over every instance's latest snapshot.
    pub subscribers: i64,
}

/// One feed's page: what it says it is, and its latest items.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FeedPage {
    #[serde(flatten)]
    pub listing: Listing,
    /// The feed's own text, HTML as often as not, as it was sent: whoever shows it sanitizes it.
    pub description: Option<String>,
    /// Why its last check failed, in words, when it did.
    pub failing: Option<String>,
    pub items: Vec<Item>,
}

/// One of a feed's latest items: what it is, never its files.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Item {
    pub guid: String,
    pub title: Option<String>,
    pub link: Option<String>,
    pub published: Option<i64>,
    pub duration: Option<i64>,
    /// As the feed sent it, unsanitized.
    pub description: Option<String>,
}

/// One feed in a snapshot: its address, and how many people on the sending server subscribe.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SnapshotEntry {
    pub url: String,
    pub subscribers: i64,
    /// The category the sending server's admin gave a feed that names none itself: used only
    /// while the feed names none.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub category: Option<String>,
}

#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)]
pub struct SnapshotResult {
    pub accepted: usize,
    /// Addresses that look like they carry a subscriber's own access.
    pub refused_private: usize,
    /// Feeds an admin removed from the directory.
    pub refused_blocked: usize,
    /// Not an http(s) address.
    pub refused_invalid: usize,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Me {
    pub name: String,
    pub access: Access,
}

/// A token whose snapshot the directory holds: how many feeds it gave, and how many people.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Instance {
    pub name: String,
    pub feeds: i64,
    pub subscribers: i64,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Blocked {
    pub id: String,
    pub url: Option<String>,
    pub title: Option<String>,
    pub at: i64,
}

/// The most a snapshot may list: far more than one server's subscriptions, few enough that one
/// request cannot hold the database for long.
pub const MAX_SNAPSHOT: usize = 20_000;

/// How often each feed is fetched again. Metadata changes slowly, and every server subscribed
/// to a feed fetches it itself anyway: every six hours was 3.9 GB of feeds a day (#243).
pub const REFRESH_EVERY: i64 = 12 * 3600;

pub struct Directory {
    pub store: Store,
    tokens: RwLock<Vec<Token>>,
    client: reqwest::Client,
    /// Wakes the refresher, so a feed a snapshot brings in is fetched now, not in a minute.
    wake: tokio::sync::Notify,
}

pub type Shared = Arc<Directory>;

impl Directory {
    pub fn new(store: Store, tokens: Vec<Token>) -> Result<Shared, reqwest::Error> {
        // No redirects: ipx_feed::fetch follows them itself, so it sees each hop.
        let client = reqwest::Client::builder()
            .connect_timeout(std::time::Duration::from_secs(10))
            .read_timeout(ipx_feed::FEED_SILENCE)
            .redirect(reqwest::redirect::Policy::none())
            .user_agent(concat!("ipx-directory/", env!("CARGO_PKG_VERSION")))
            .build()?;
        Ok(Arc::new(Self { store, tokens: RwLock::new(tokens), client, wake: tokio::sync::Notify::new() }))
    }

    /// Takes the tokens of a config file read again, and drops the snapshots of any that went.
    pub async fn set_tokens(&self, tokens: Vec<Token>) -> anyhow::Result<u64> {
        let names: Vec<String> = tokens.iter().map(|t| t.name.clone()).collect();
        *self.tokens.write().unwrap() = tokens;
        self.store.keep_instances(&names).await
    }

    fn caller(&self, presented: &str) -> Option<Me> {
        let h = config::hash(presented);
        // Every token compared, whether or not one matched, so timing says nothing either way.
        self.tokens
            .read()
            .unwrap()
            .iter()
            .fold(None, |found, t| if config::same_secret(&t.sha256, &h) { Some(Me { name: t.name.clone(), access: t.access }) } else { found })
    }

    /// Fetches one feed and stores what it found.
    pub async fn refresh(&self, due: &store::Due) -> anyhow::Result<()> {
        let got = match ipx_feed::fetch(&self.client, &due.url, None, due.etag.as_deref(), due.last_modified.as_deref()).await {
            Ok((ipx_feed::Fetched::NotModified, _)) => Refreshed::NotModified,
            Ok((ipx_feed::Fetched::Body { bytes, etag, last_modified }, _)) => {
                if ipx_feed::is_opml(&bytes) {
                    // ipx sends the feeds inside an OPML, never the OPML.
                    Refreshed::Failed("an OPML list, not a feed".into())
                } else {
                    match ipx_feed::parse(&bytes) {
                        Ok(parsed) => Refreshed::Feed { parsed, etag, last_modified },
                        Err(e) => Refreshed::Failed(format!("{e:#}")),
                    }
                }
            }
            Err(e) => Refreshed::Failed(format!("{e:#}")),
        };
        if let Refreshed::Failed(e) = &got {
            tracing::info!(feed = %due.id, error = %ipx_feed::redact_text(e), "feed check failed");
        }
        // ponytail: a feed that moved for good is still asked for at its old address, which
        // redirects; follow the move when a feed's id can change without losing its counts.
        self.store.save(&due.id, got).await
    }
}

/// Fetches whatever is due, a few at a time, forever.
pub async fn refresher(dir: Shared) {
    loop {
        match dir.store.due(REFRESH_EVERY, 25).await {
            Ok(due) => {
                if !due.is_empty() {
                    // A round is a trace, each feed's check a span in it (#213).
                    let round = tracing::info_span!("refresh", feeds = due.len());
                    tracing::Instrument::instrument(
                        async {
                            // A few at a time: in turn, two publishers timing out held a round
                            // for a minute and more (#227).
                            futures_util::StreamExt::for_each_concurrent(futures_util::stream::iter(&due), 5, |d| {
                                let check = tracing::info_span!("feed", feed = %d.id);
                                async {
                                    if let Err(e) = tracing::Instrument::instrument(dir.refresh(d), check).await {
                                        tracing::warn!(feed = %d.id, error = %format!("{e:#}"), "could not store a feed check");
                                    }
                                }
                            })
                            .await;
                            // One line a round, for the dashboard's refresher panel (#210); failures have their own.
                            tracing::info!(checked = due.len(), "feeds checked");
                        },
                        round,
                    )
                    .await;
                    continue;
                }
            }
            Err(e) => tracing::warn!(error = %format!("{e:#}"), "could not read which feeds are due"),
        }
        tokio::select! {
            _ = tokio::time::sleep(Duration::from_secs(60)) => {}
            _ = dir.wake.notified() => {}
        }
    }
}

/// Reads the config file again whenever it changes, checked each minute. A file that fails to
/// load leaves the tokens as they were, so a typo locks nobody out.
pub async fn reloader(dir: Shared, path: std::path::PathBuf) {
    let mtime = |p: &std::path::Path| std::fs::metadata(p).and_then(|m| m.modified()).ok();
    let mut last = mtime(&path);
    loop {
        tokio::time::sleep(Duration::from_secs(60)).await;
        let now = mtime(&path);
        if now == last {
            continue;
        }
        last = now;
        match config::Config::load(&path).and_then(|c| c.tokens()) {
            Ok(tokens) => {
                let n = tokens.len();
                match dir.set_tokens(tokens).await {
                    Ok(dropped) => tracing::info!(tokens = n, dropped, "config read again"),
                    Err(e) => tracing::warn!(error = %format!("{e:#}"), "config read again, but old snapshots could not be dropped"),
                }
            }
            Err(e) => tracing::warn!(error = %format!("{e:#}"), "config changed but does not load; keeping the tokens as they were"),
        }
    }
}

pub fn router(dir: Shared) -> Router {
    let v1 = Router::new()
        .route("/v1/me", get(me))
        .route("/v1/feeds", get(list))
        .route("/v1/popular", get(popular))
        .route("/v1/feeds/{id}", get(feed).delete(remove))
        .route("/v1/feeds/{id}/refresh", post(refresh))
        .route("/v1/snapshot", post(snapshot))
        .route("/v1/blocked", get(blocked))
        .route("/v1/blocked/{id}", delete(unblock))
        .route("/v1/instances", get(instances))
        .route("/v1/instances/{name}", delete(drop_instance))
        .layer(axum::middleware::from_fn_with_state(dir.clone(), authenticate));
    Router::new()
        .route("/healthz", get(healthz))
        .merge(v1)
        // Anyone may browse, no token, read only (#215).
        .merge(site::router())
        // Brotli, zstd or gzip as the caller takes them: the feed list went to ipx as 603 KB of
        // JSON, most of a Directory page's time (#253). Inside the access log, which records what
        // was sent.
        .layer(tower_http::compression::CompressionLayer::new().br(true).zstd(true).gzip(true))
        // And takes a request's body gzipped: ipx sends its snapshot so, about 100 KB as JSON.
        .layer(tower_http::decompression::RequestDecompressionLayer::new().gzip(true))
        .layer(axum::middleware::from_fn(access_log))
        .with_state(dir)
}

struct ApiError(StatusCode, String);

impl IntoResponse for ApiError {
    fn into_response(self) -> Response {
        (self.0, Json(serde_json::json!({ "error": self.1 }))).into_response()
    }
}

impl From<anyhow::Error> for ApiError {
    fn from(e: anyhow::Error) -> Self {
        tracing::error!(error = %format!("{e:#}"), "request failed");
        ApiError(StatusCode::INTERNAL_SERVER_ERROR, "something went wrong; the directory's log says what".into())
    }
}

/// Who made the request, for the access log, set by `authenticate`.
#[derive(Clone)]
struct Who(String);

/// One line a request, with the token's name and never the token: what the Grafana panels read.
async fn access_log(req: Request, next: Next) -> Response {
    let (method, path) = (req.method().clone(), req.uri().path().to_owned());
    let start = Instant::now();
    // A span for each request but the healthcheck's, a child of the caller's when it sent a
    // traceparent, as ipx does on its calls here (#213).
    let span = if path == "/healthz" {
        tracing::Span::none()
    } else {
        tracing::info_span!(
            "http",
            otel.name = %format!("{method} {path}"),
            http.request.method = %method,
            url.path = %path,
            http.response.status_code = tracing::field::Empty,
        )
    };
    if let Some(parent) = otel::remote_parent(req.headers(), None) {
        use tracing_opentelemetry::OpenTelemetrySpanExt;
        let _ = span.set_parent(parent);
    }
    let resp = tracing::Instrument::instrument(next.run(req), span.clone()).await;
    span.record("http.response.status_code", resp.status().as_u16());
    let _in = span.enter();
    let who = resp.extensions().get::<Who>().map_or("", |w| w.0.as_str()).to_owned();
    let status = resp.status().as_u16();
    if path != "/healthz" {
        tracing::info!(
            target: "ipx_directory::http",
            method = %method, path = %path, status, duration_ms = start.elapsed().as_millis() as u64, token = %who,
            "{method} {path} -> {status}"
        );
    }
    resp
}

async fn authenticate(State(dir): State<Shared>, mut req: Request, next: Next) -> Response {
    let presented = req
        .headers()
        .get(header::AUTHORIZATION)
        .and_then(|v| v.to_str().ok())
        .and_then(|v| v.strip_prefix("Bearer "))
        .map(str::trim)
        .filter(|t| !t.is_empty());
    let Some(me) = presented.and_then(|t| dir.caller(t)) else {
        return ApiError(StatusCode::UNAUTHORIZED, "a valid token is needed: Authorization: Bearer <token>".into()).into_response();
    };
    let who = Who(me.name.clone());
    req.extensions_mut().insert(me);
    let mut resp = next.run(req).await;
    resp.extensions_mut().insert(who);
    resp
}

fn need(me: &Me, access: Access) -> Result<(), ApiError> {
    if me.access >= access {
        Ok(())
    } else {
        let what = match access {
            Access::Ro => "browse",
            Access::Rw => "send snapshots",
            Access::Rwa => "administer the directory",
        };
        Err(ApiError(StatusCode::FORBIDDEN, format!("this token may not {what}")))
    }
}

async fn healthz(State(dir): State<Shared>) -> Result<&'static str, ApiError> {
    dir.store.ping().await?;
    Ok("ok")
}

async fn me(Extension(me): Extension<Me>) -> Json<Me> {
    Json(me)
}

async fn list(State(dir): State<Shared>) -> Result<Json<Vec<Listing>>, ApiError> {
    Ok(Json(dir.store.list().await?))
}

#[derive(Deserialize)]
struct Limit {
    limit: Option<u64>,
}

async fn popular(State(dir): State<Shared>, Query(q): Query<Limit>) -> Result<Json<Vec<Listing>>, ApiError> {
    Ok(Json(dir.store.popular(q.limit.unwrap_or(10).clamp(1, 100)).await?))
}

async fn feed(State(dir): State<Shared>, Path(id): Path<String>) -> Result<Json<FeedPage>, ApiError> {
    dir.store.feed(&id).await?.map(Json).ok_or_else(|| ApiError(StatusCode::NOT_FOUND, format!("{id:?} is not in the directory")))
}

async fn snapshot(
    State(dir): State<Shared>,
    Extension(me): Extension<Me>,
    Json(entries): Json<Vec<SnapshotEntry>>,
) -> Result<Json<SnapshotResult>, ApiError> {
    need(&me, Access::Rw)?;
    if entries.len() > MAX_SNAPSHOT {
        return Err(ApiError(StatusCode::PAYLOAD_TOO_LARGE, format!("a snapshot lists at most {MAX_SNAPSHOT} feeds")));
    }
    let r = dir.store.snapshot(&me.name, &entries).await?;
    dir.wake.notify_one();
    tracing::info!(
        instance = %me.name, accepted = r.accepted, refused_private = r.refused_private,
        refused_blocked = r.refused_blocked, refused_invalid = r.refused_invalid, "snapshot"
    );
    Ok(Json(r))
}

async fn remove(State(dir): State<Shared>, Extension(me): Extension<Me>, Path(id): Path<String>) -> Result<StatusCode, ApiError> {
    need(&me, Access::Rwa)?;
    dir.store.block(&id).await?;
    tracing::info!(feed = %id, by = %me.name, "feed removed and blocked");
    Ok(StatusCode::NO_CONTENT)
}

async fn blocked(State(dir): State<Shared>, Extension(me): Extension<Me>) -> Result<Json<Vec<Blocked>>, ApiError> {
    need(&me, Access::Rwa)?;
    Ok(Json(dir.store.blocked().await?))
}

async fn unblock(State(dir): State<Shared>, Extension(me): Extension<Me>, Path(id): Path<String>) -> Result<StatusCode, ApiError> {
    need(&me, Access::Rwa)?;
    if !dir.store.unblock(&id).await? {
        return Err(ApiError(StatusCode::NOT_FOUND, format!("{id:?} is not blocked")));
    }
    tracing::info!(feed = %id, by = %me.name, "feed unblocked");
    Ok(StatusCode::NO_CONTENT)
}

async fn instances(State(dir): State<Shared>, Extension(me): Extension<Me>) -> Result<Json<Vec<Instance>>, ApiError> {
    need(&me, Access::Rwa)?;
    Ok(Json(dir.store.instances().await?))
}

async fn drop_instance(State(dir): State<Shared>, Extension(me): Extension<Me>, Path(name): Path<String>) -> Result<StatusCode, ApiError> {
    need(&me, Access::Rwa)?;
    let n = dir.store.drop_instance(&name).await?;
    tracing::info!(instance = %name, feeds = n, by = %me.name, "snapshot dropped");
    Ok(StatusCode::NO_CONTENT)
}

async fn refresh(State(dir): State<Shared>, Extension(me): Extension<Me>, Path(id): Path<String>) -> Result<Json<FeedPage>, ApiError> {
    need(&me, Access::Rwa)?;
    let due = dir.store.due_one(&id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, format!("{id:?} is not in the directory")))?;
    dir.refresh(&due).await?;
    feed(State(dir), Path(id)).await
}

#[cfg(test)]
mod tests {
    use super::*;
    use axum::body::Body;
    use tower::ServiceExt;

    async fn app() -> (Router, Shared) {
        let tokens = ["ro", "rw", "rwa"]
            .iter()
            .map(|n| Token {
                name: format!("{n}-token"),
                access: serde_json::from_value(serde_json::json!(n)).unwrap(),
                sha256: config::hash(&format!("secret-{n}")),
            })
            .collect();
        let dir = Directory::new(Store::scratch().await, tokens).unwrap();
        (router(dir.clone()), dir)
    }

    async fn call(app: &Router, method: &str, path: &str, token: Option<&str>, body: Option<serde_json::Value>) -> (u16, serde_json::Value) {
        let mut req = axum::http::Request::builder().method(method).uri(path);
        if let Some(t) = token {
            req = req.header("authorization", format!("Bearer secret-{t}"));
        }
        let req = match body {
            Some(b) => req.header("content-type", "application/json").body(Body::from(b.to_string())),
            None => req.body(Body::empty()),
        }
        .unwrap();
        let resp = app.clone().oneshot(req).await.unwrap();
        let status = resp.status().as_u16();
        let bytes = axum::body::to_bytes(resp.into_body(), 1 << 20).await.unwrap();
        (status, serde_json::from_slice(&bytes).unwrap_or(serde_json::Value::Null))
    }

    #[tokio::test]
    async fn a_snapshot_may_come_gzipped() {
        use std::io::Write as _;
        let (app, _) = app().await;
        let json = serde_json::json!([{"url": "https://example.com/feed.xml", "subscribers": 2}]).to_string();
        let mut gz = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
        gz.write_all(json.as_bytes()).unwrap();
        // As ipx sends it (#253).
        let req = axum::http::Request::builder()
            .method("POST")
            .uri("/v1/snapshot")
            .header("authorization", "Bearer secret-rw")
            .header("content-type", "application/json")
            .header("content-encoding", "gzip")
            .body(Body::from(gz.finish().unwrap()))
            .unwrap();
        let resp = app.clone().oneshot(req).await.unwrap();
        assert_eq!(resp.status(), 200);
        let got: serde_json::Value = serde_json::from_slice(&axum::body::to_bytes(resp.into_body(), 1 << 20).await.unwrap()).unwrap();
        assert_eq!(got["accepted"], 1);
    }

    #[tokio::test]
    async fn answers_come_compressed_as_the_caller_asks() {
        let (app, _) = app().await;
        for (asked, sent) in [("br", Some("br")), ("zstd", Some("zstd")), ("gzip", Some("gzip")), ("identity", None)] {
            // A page, not an empty feed list: under 32 bytes is sent as it is.
            let req = axum::http::Request::builder()
                .uri("/")
                .header("accept-encoding", asked)
                .body(Body::empty())
                .unwrap();
            let resp = app.clone().oneshot(req).await.unwrap();
            assert_eq!(resp.status(), 200);
            let got = resp.headers().get("content-encoding").map(|v| v.to_str().unwrap().to_owned());
            assert_eq!(got.as_deref(), sent, "asked for {asked} (#253)");
        }
    }

    async fn get_page(app: &Router, path: &str) -> (u16, String, Option<String>) {
        let req = axum::http::Request::builder().uri(path).body(Body::empty()).unwrap();
        let resp = app.clone().oneshot(req).await.unwrap();
        let status = resp.status().as_u16();
        let csp = resp.headers().get("content-security-policy").map(|v| v.to_str().unwrap().to_owned());
        let bytes = axum::body::to_bytes(resp.into_body(), 1 << 22).await.unwrap();
        (status, String::from_utf8(bytes.to_vec()).unwrap(), csp)
    }

    #[tokio::test]
    async fn anyone_browses_the_pages_and_the_api_still_needs_a_token() {
        let (app, dir) = app().await;
        let snap = serde_json::json!([{ "url": "https://x.test/a.xml", "subscribers": 3 }]);
        call(&app, "POST", "/v1/snapshot", Some("rw"), Some(snap)).await;
        store::tests::fetch_all(&dir.store).await;
        let id = dir.store.list().await.unwrap()[0].id.clone();

        let (s, home, csp) = get_page(&app, "/").await;
        assert_eq!(s, 200);
        assert!(csp.unwrap().contains("script-src 'self'"));
        assert!(home.contains("Most subscribed") && home.contains(&format!("/feed/{id}")), "{home}");
        assert!(get_page(&app, "/all").await.1.contains("3 people subscribe"));
        let (s, page, _) = get_page(&app, &format!("/feed/{id}")).await;
        assert_eq!(s, 200);
        // Its address, to subscribe to anywhere.
        assert!(page.contains("value=\"https://x.test/a.xml\""), "{page}");
        assert!(!page.contains("<script>"), "the publisher's HTML is text here");
        assert_eq!(get_page(&app, "/feed/nope").await.0, 404);
        assert_eq!(get_page(&app, "/category/Nothing").await.0, 404);
        assert_eq!(get_page(&app, "/search?q=").await.0, 303);
        assert_eq!(get_page(&app, "/site.js").await.0, 200);
        // Reading the pages takes no token; the API still does.
        assert_eq!(call(&app, "GET", "/v1/feeds", None, None).await.0, 401);
        assert_eq!(call(&app, "DELETE", &format!("/v1/feeds/{id}"), None, None).await.0, 401);
        assert_eq!(call(&app, "DELETE", &format!("/feed/{id}"), None, None).await.0, 405);
    }

    #[tokio::test]
    async fn each_access_level_does_what_it_may_and_no_more() {
        let (app, dir) = app().await;
        let snap = serde_json::json!([{ "url": "https://x.test/a.xml", "subscribers": 2, "category": "Comedy" }]);

        assert_eq!(call(&app, "GET", "/v1/feeds", None, None).await.0, 401);
        assert_eq!(call(&app, "GET", "/v1/feeds", Some("wrong"), None).await.0, 401);
        assert_eq!(call(&app, "GET", "/healthz", None, None).await.0, 200);

        let (s, me) = call(&app, "GET", "/v1/me", Some("ro"), None).await;
        assert_eq!((s, me["name"].as_str(), me["access"].as_str()), (200, Some("ro-token"), Some("ro")));
        assert_eq!(call(&app, "POST", "/v1/snapshot", Some("ro"), Some(snap.clone())).await.0, 403);

        let (s, r) = call(&app, "POST", "/v1/snapshot", Some("rw"), Some(snap.clone())).await;
        assert_eq!((s, r["accepted"].as_u64()), (200, Some(1)));
        store::tests::fetch_all(&dir.store).await;
        let (_, list) = call(&app, "GET", "/v1/feeds", Some("ro"), None).await;
        let id = list[0]["id"].as_str().unwrap().to_owned();
        assert_eq!(list[0]["subscribers"], 2);
        assert_ne!(list[0]["category"], "Comedy", "the feed's own category wins over the sender's");
        assert_eq!(call(&app, "GET", &format!("/v1/feeds/{id}"), Some("ro"), None).await.0, 200);
        assert_eq!(call(&app, "GET", "/v1/feeds/nope", Some("ro"), None).await.0, 404);
        assert_eq!(call(&app, "GET", "/v1/popular?limit=5", Some("ro"), None).await.1[0]["id"], id.as_str());

        for (m, p) in [("DELETE", format!("/v1/feeds/{id}")), ("GET", "/v1/blocked".into()), ("DELETE", "/v1/instances/rw-token".into())] {
            assert_eq!(call(&app, m, &p, Some("rw"), None).await.0, 403, "{m} {p}");
        }
        assert_eq!(call(&app, "DELETE", &format!("/v1/feeds/{id}"), Some("rwa"), None).await.0, 204);
        assert_eq!(call(&app, "GET", "/v1/blocked", Some("rwa"), None).await.1[0]["id"], id.as_str());
        assert_eq!(call(&app, "POST", "/v1/snapshot", Some("rw"), Some(snap.clone())).await.1["refused_blocked"], 1);
        assert_eq!(call(&app, "DELETE", &format!("/v1/blocked/{id}"), Some("rwa"), None).await.0, 204);
        assert_eq!(call(&app, "DELETE", &format!("/v1/blocked/{id}"), Some("rwa"), None).await.0, 404);
        call(&app, "POST", "/v1/snapshot", Some("rw"), Some(snap)).await;
        assert_eq!(call(&app, "DELETE", "/v1/instances/rw-token", Some("rwa"), None).await.0, 204);
        assert_eq!(call(&app, "GET", "/v1/feeds", Some("ro"), None).await.1, serde_json::json!([]));
    }

    #[tokio::test]
    async fn a_revoked_token_loses_its_access_and_its_counts() {
        let dir = Directory::new(Store::scratch().await, vec![]).unwrap();
        let t = |n: &str| Token { name: n.into(), access: Access::Rw, sha256: config::hash(&format!("secret-{n}")) };
        dir.set_tokens(vec![t("a"), t("b")]).await.unwrap();
        let app = router(dir.clone());
        let snap = serde_json::json!([{ "url": "https://x.test/a.xml", "subscribers": 1 }]);
        call(&app, "POST", "/v1/snapshot", Some("a"), Some(snap.clone())).await;
        call(&app, "POST", "/v1/snapshot", Some("b"), Some(snap)).await;
        store::tests::fetch_all(&dir.store).await;
        assert_eq!(call(&app, "GET", "/v1/feeds", Some("a"), None).await.1[0]["subscribers"], 2);

        dir.set_tokens(vec![t("a")]).await.unwrap();
        assert_eq!(call(&app, "GET", "/v1/feeds", Some("b"), None).await.0, 401);
        assert_eq!(call(&app, "GET", "/v1/feeds", Some("a"), None).await.1[0]["subscribers"], 1);
    }
}