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/directory/src/lib.rs

@@ -94,6 +94,14 @@ pub struct Me {
     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,
@@ -242,6 +250,7 @@ pub fn router(dir: Shared) -> Router {
         .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()
@@ -402,6 +411,11 @@ async fn unblock(State(dir): State<Shared>, Extension(me): Extension<Me>, Path(i
     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?;

crates/directory/src/store.rs

@@ -5,7 +5,7 @@
 use anyhow::Result;
 use sea_orm::{ConnectionTrait, DatabaseConnection, Statement, TransactionTrait, Value};
 
-use crate::{Blocked, FeedPage, Item, Listing, SnapshotEntry, SnapshotResult};
+use crate::{Blocked, FeedPage, Instance, Item, Listing, SnapshotEntry, SnapshotResult};
 
 /// A feed's items as its page shows them, and as many as ipx's own Directory page showed.
 pub const ITEMS: usize = 20;
@@ -334,6 +334,19 @@ impl Store {
         Ok(self.exec("DELETE FROM blocked WHERE id = $1", vec![id.into()]).await? > 0)
     }
 
+    pub async fn instances(&self) -> Result<Vec<Instance>> {
+        // Postgres sums a BIGINT as NUMERIC, which does not read as an i64.
+        self.rows(
+            "SELECT instance, count(*) AS feeds, CAST(sum(subscribers) AS BIGINT) AS subscribers
+             FROM snapshots GROUP BY instance ORDER BY instance",
+            vec![],
+        )
+        .await?
+        .iter()
+        .map(|r| Ok(Instance { name: r.try_get("", "instance")?, feeds: r.try_get("", "feeds")?, subscribers: r.try_get("", "subscribers")? }))
+        .collect()
+    }
+
     pub async fn blocked(&self) -> Result<Vec<Blocked>> {
         self.rows("SELECT id, url, title, at FROM blocked ORDER BY at DESC, id", vec![])
             .await?
@@ -498,6 +511,9 @@ pub(crate) mod tests {
         assert_eq!(all.len(), 1);
         assert_eq!(all[0].subscribers, 3);
 
+        let i = s.instances().await.unwrap();
+        assert_eq!(i.iter().map(|i| (i.name.as_str(), i.feeds, i.subscribers)).collect::<Vec<_>>(), [("a", 1, 1), ("b", 1, 2)]);
+
         // A token taken out of the config file takes its counts.
         s.keep_instances(&["b".into()]).await.unwrap();
         assert_eq!(s.list().await.unwrap()[0].subscribers, 2);

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