Administer the directory from iPX with an rwa token (#206)

A Directory feed's page offers Check now and Remove to an admin whose token is rwa (the page
learns it from can_admin on the feed's page, asked of the directory only for admins). Admin,
Directory lists the servers sharing, from a new GET /v1/instances, and the blocked feeds by
title: a feed is often blocked because its address carries a key, so the address stays in the
directory. Unblocking or dropping a snapshot sends this server's snapshot again at once; since
directory's 403 reaches the page as 403, not 502.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

crates/ipx/CHANGELOG.md

@@ -14,6 +14,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- With a directory token that may administer it, ipx's admins remove a feed from the directory (blocking it) or check it now, list and unblock what is blocked, and list and drop the servers sharing (#206).
 - ipx passes its trace on when it calls the directory, so a Directory page is one trace from the browser to the directory (#213).
 
 ### Changed

crates/ipx/src/directory.rs

@@ -71,7 +71,13 @@ async fn call<T: DeserializeOwned>(ctx: &Ctx, method: reqwest::Method, path: &st
         let msg = resp.json::<serde_json::Value>().await.ok().and_then(|v| v["error"].as_str().map(str::to_owned)).unwrap_or_default();
         return Err(DirError::Refused(status.as_u16(), msg));
     }
-    resp.json().await.map_err(|e| DirError::Unreachable(format!("its answer did not read: {e:#}")))
+    // A removal or an unblock answers 204, with nothing to read: that reads as null, which is ().
+    let body = resp.bytes().await.map_err(|e| DirError::Unreachable(format!("{e:#}")))?;
+    serde_json::from_slice(if body.is_empty() { b"null" } else { &body }).map_err(|e| DirError::Unreachable(format!("its answer did not read: {e:#}")))
+}
+
+fn seg(s: &str) -> percent_encoding::PercentEncode<'_> {
+    percent_encoding::utf8_percent_encode(s, percent_encoding::NON_ALPHANUMERIC)
 }
 
 pub async fn me(ctx: &Ctx) -> Result<Me, DirError> {
@@ -87,7 +93,47 @@ pub async fn popular(ctx: &Ctx) -> Result<Vec<Listing>, DirError> {
 }
 
 pub async fn page(ctx: &Ctx, id: &str) -> Result<FeedPage, DirError> {
-    call(ctx, reqwest::Method::GET, &format!("/v1/feeds/{}", percent_encoding::utf8_percent_encode(id, percent_encoding::NON_ALPHANUMERIC)), None).await
+    call(ctx, reqwest::Method::GET, &format!("/v1/feeds/{}", seg(id)), None).await
+}
+
+// Administering the directory (#206): each needs an rwa token, which the directory checks.
+
+/// Removes a feed and blocks it, so no snapshot brings it back.
+pub async fn remove(ctx: &Ctx, id: &str) -> Result<(), DirError> {
+    call(ctx, reqwest::Method::DELETE, &format!("/v1/feeds/{}", seg(id)), None).await
+}
+
+pub async fn refresh(ctx: &Ctx, id: &str) -> Result<serde::de::IgnoredAny, DirError> {
+    call(ctx, reqwest::Method::POST, &format!("/v1/feeds/{}/refresh", seg(id)), None).await
+}
+
+pub async fn blocked(ctx: &Ctx) -> Result<Vec<ipx_directory::Blocked>, DirError> {
+    call(ctx, reqwest::Method::GET, "/v1/blocked", None).await
+}
+
+pub async fn unblock(ctx: &Ctx, id: &str) -> Result<(), DirError> {
+    call(ctx, reqwest::Method::DELETE, &format!("/v1/blocked/{}", seg(id)), None).await
+}
+
+pub async fn instances(ctx: &Ctx) -> Result<Vec<ipx_directory::Instance>, DirError> {
+    call(ctx, reqwest::Method::GET, "/v1/instances", None).await
+}
+
+/// Drops one server's snapshot, and its counts with it. It comes back with that server's next
+/// one: to keep a server out, revoke its token.
+pub async fn drop_instance(ctx: &Ctx, name: &str) -> Result<(), DirError> {
+    call(ctx, reqwest::Method::DELETE, &format!("/v1/instances/{}", seg(name)), None).await
+}
+
+/// Has the snapshot sent again now, changed or not.
+pub fn resend_now(ctx: &Ctx) {
+    ctx.directory_resend.store(true, std::sync::atomic::Ordering::Relaxed);
+    ctx.directory_wake.notify_one();
+}
+
+/// Whether this server's token may administer the directory.
+pub async fn can_admin(ctx: &Ctx) -> bool {
+    me(ctx).await.is_ok_and(|m| m.access == ipx_directory::config::Access::Rwa)
 }
 
 /// Whether a feed is someone's own: marked private, given a username or password, or at an
@@ -150,7 +196,8 @@ pub async fn run(ctx: std::sync::Arc<Ctx>, mut stop: tokio::sync::watch::Receive
             let got = match snapshot(&ctx).await {
                 Ok(entries) => {
                     let key = serde_json::to_vec(&(target(&cfg), &entries)).unwrap_or_default();
-                    if resend(last.as_ref(), &key) {
+                    let forced = ctx.directory_resend.swap(false, std::sync::atomic::Ordering::Relaxed);
+                    if forced || resend(last.as_ref(), &key) {
                         let span = tracing::info_span!(target: "ipx::directory", "directory_share", feeds = entries.len());
                         let got = tracing::Instrument::instrument(share(&ctx, &entries), span).await;
                         if got.is_ok() {

crates/ipx/src/main.rs

@@ -166,6 +166,9 @@ pub struct Ctx {
     pub directory_wake: tokio::sync::Notify,
     /// When the snapshot last went and what came of it, for the admin page.
     pub directory_shared: std::sync::Mutex<Option<directory::Shared>>,
+    /// Send the snapshot even if it has not changed: after an unblock or a dropped snapshot the
+    /// directory needs it again, though nothing here differs (#206).
+    pub directory_resend: std::sync::atomic::AtomicBool,
 }
 
 /// A copy of the config to change, holding off every other change until `store_cfg` keeps it.
@@ -365,6 +368,7 @@ async fn main() -> Result<()> {
         deadair: Default::default(),
         directory_wake: Default::default(),
         directory_shared: Default::default(),
+        directory_resend: Default::default(),
     });
 
     let result = match cli.command {
@@ -2501,6 +2505,7 @@ mod tests {
             deadair: Default::default(),
             directory_wake: Default::default(),
             directory_shared: Default::default(),
+            directory_resend: Default::default(),
             directory_cleaned: Default::default(),
         }
     }

crates/ipx/src/web.rs

@@ -62,7 +62,11 @@ pub fn router(state: WebState) -> Router {
         .route("/api/status", get(status))
         .route("/api/popular", get(get_popular))
         .route("/api/directory", get(get_directory))
-        .route("/api/directory/{id}", get(get_listed))
+        .route("/api/directory/{id}", get(get_listed).delete(remove_listed))
+        .route("/api/directory/{id}/refresh", post(refresh_listed))
+        .route("/api/directory-admin", get(get_directory_admin))
+        .route("/api/directory-blocked/{id}", delete(unblock_listed))
+        .route("/api/directory-instances/{name}", delete(drop_instance))
         .route("/api/popular/{id}", post(subscribe_popular))
         .route("/api/users", get(list_users).post(add_user))
         .route("/api/users/{id}", patch(patch_user).delete(remove_user))
@@ -814,6 +818,8 @@ fn dir_error(e: crate::directory::DirError) -> ApiError {
     let status = match &e {
         DirError::NotConfigured => StatusCode::SERVICE_UNAVAILABLE,
         DirError::Refused(404, _) => StatusCode::NOT_FOUND,
+        // This server's token may not do it: the admin asked for something only rwa may (#206).
+        DirError::Refused(403, _) => StatusCode::FORBIDDEN,
         _ => StatusCode::BAD_GATEWAY,
     };
     if !matches!(e, DirError::NotConfigured | DirError::Refused(404, _)) {
@@ -888,16 +894,21 @@ struct Listed {
     /// Why its last check failed, when it did (#142), in the directory's words.
     failing: Option<String>,
     items: Vec<ListedItem>,
+    /// An admin here whose token may administer the directory, who sees Remove and Refresh (#206).
+    can_admin: bool,
 }
 
 /// A listed feed's description and latest items, from the directory, sanitized here: they are
 /// the feed's own HTML, which the directory passes on as it was sent.
 async fn get_listed(
     State(state): State<WebState>,
-    _user: crate::db::User,
+    user: crate::db::User,
     Path(id): Path<String>,
 ) -> Result<Json<Listed>, ApiError> {
-    let page = crate::directory::page(&state.ctx, &id).await.map_err(dir_error)?;
+    let (page, can_admin) = tokio::join!(crate::directory::page(&state.ctx, &id), async {
+        user.is_admin && crate::directory::can_admin(&state.ctx).await
+    });
+    let page = page.map_err(dir_error)?;
     let mut sanitizer = feed_sanitizer();
     let description = page.description.as_deref().map(|d| clean_description(&mut sanitizer, d, None));
     let items = page
@@ -912,7 +923,60 @@ async fn get_listed(
             duration: e.duration,
         })
         .collect();
-    Ok(Json(Listed { description, failing: page.failing, items }))
+    Ok(Json(Listed { description, failing: page.failing, items, can_admin }))
+}
+
+// ---- administering the directory (#206): an admin here, and a token there that may ----
+
+fn admin_only(user: &crate::db::User) -> Result<(), ApiError> {
+    if user.is_admin { Ok(()) } else { Err(ApiError::forbidden("only an admin administers the directory")) }
+}
+
+async fn remove_listed(State(state): State<WebState>, user: crate::db::User, Path(id): Path<String>) -> Result<StatusCode, ApiError> {
+    admin_only(&user)?;
+    crate::directory::remove(&state.ctx, &id).await.map_err(dir_error)?;
+    tracing::info!(by = %user.name, feed = %id, "removed a feed from the directory");
+    Ok(StatusCode::NO_CONTENT)
+}
+
+async fn refresh_listed(State(state): State<WebState>, user: crate::db::User, Path(id): Path<String>) -> Result<StatusCode, ApiError> {
+    admin_only(&user)?;
+    crate::directory::refresh(&state.ctx, &id).await.map_err(dir_error)?;
+    Ok(StatusCode::NO_CONTENT)
+}
+
+/// A blocked feed, by its title: a feed is often blocked because its address holds someone's
+/// key, so the address stays in the directory.
+#[derive(Serialize)]
+struct BlockedRow {
+    id: String,
+    title: Option<String>,
+    at: i64,
+}
+
+async fn get_directory_admin(State(state): State<WebState>, user: crate::db::User) -> Result<Json<serde_json::Value>, ApiError> {
+    admin_only(&user)?;
+    let (instances, blocked) = tokio::join!(crate::directory::instances(&state.ctx), crate::directory::blocked(&state.ctx));
+    let blocked: Vec<BlockedRow> = blocked.map_err(dir_error)?.into_iter().map(|b| BlockedRow { id: b.id, title: b.title, at: b.at }).collect();
+    Ok(Json(serde_json::json!({ "instances": instances.map_err(dir_error)?, "blocked": blocked })))
+}
+
+async fn unblock_listed(State(state): State<WebState>, user: crate::db::User, Path(id): Path<String>) -> Result<StatusCode, ApiError> {
+    admin_only(&user)?;
+    crate::directory::unblock(&state.ctx, &id).await.map_err(dir_error)?;
+    // Unblocked, it comes back with a snapshot that has it, and this one has not changed.
+    crate::directory::resend_now(&state.ctx);
+    tracing::info!(by = %user.name, feed = %id, "unblocked a feed in the directory");
+    Ok(StatusCode::NO_CONTENT)
+}
+
+async fn drop_instance(State(state): State<WebState>, user: crate::db::User, Path(name): Path<String>) -> Result<StatusCode, ApiError> {
+    admin_only(&user)?;
+    crate::directory::drop_instance(&state.ctx, &name).await.map_err(dir_error)?;
+    // When it was this server's own, its feeds come back now rather than in half an hour.
+    crate::directory::resend_now(&state.ctx);
+    tracing::info!(by = %user.name, instance = %name, "dropped a server's snapshot from the directory");
+    Ok(StatusCode::NO_CONTENT)
 }
 
 /// Subscribes by the directory's id, because the page never sees a URL: the address comes from