rays/ipx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 18 Wiki Activity

The Directory comes from the directory service (#205)

Browse Source 
ipx asks ipx-directory on its people's behalf, with the token an admin set
in the new Admin, Directory tab, so the page never talks to it and never
sees the token or a feed's address. /api/directory, /api/popular and a
feed's Directory page answer from it; subscribing takes the address from
the directory and goes through Add feed's steps. With no directory set
up, the Directory says so. The local directory code is gone; its privacy
checks are the snapshot builder's now, and their tests moved with them.

With sharing on, ipx sends its snapshot after every scan, whenever the
config changes (set_cfg) and every half hour: each public feed and how
many here subscribe, never an OPML or Patreon creator itself or anything
private. A feed can be marked private in its settings, for a paid feed
whose address does not give it away; anyone subscribed may mark it, only
an admin unmark it. The admin's category for a feed that names none goes
with it, which the directory uses only while the feed names none, so the
category an admin gives a blog still reaches the Directory.

The directory lists only feeds it has read as a feed: the first snapshot
after a start went before the first scan, when an OPML was not yet known
to be one, and the directory listed the OPML. It now fetches what a
snapshot brings in at once instead of within the minute.

The browser suite runs its own ipx-directory on SQLite; the Directory's
tests match feeds by title, since the directory names them by address.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Ray Slakinski
2026-10-12 04:31:32 +00:00
parent ed82166745
commit c47aa57909
19 changed files with 710 additions and 246 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
crates/directory/src/lib.rs
View File

@@ -70,6 +70,10 @@ pub struct Item {
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)]
@@ -109,6 +113,8 @@ 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>;
@@ -120,7 +126,7 @@ impl Directory {
.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 }))
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.
@@ -182,7 +188,10 @@ pub async fn refresher(dir: Shared) {
}
Err(e) => tracing::warn!(error = %format!("{e:#}"), "could not read which feeds are due"),
}
tokio::time::sleep(Duration::from_secs(60)).await;
tokio::select! {
_ = tokio::time::sleep(Duration::from_secs(60)) => {}
_ = dir.wake.notified() => {}
}
}
}
@@ -333,6 +342,7 @@ async fn 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"
@@ -381,7 +391,7 @@ mod tests {
use axum::body::Body;
use tower::ServiceExt;
async fn app() -> Router {
async fn app() -> (Router, Shared) {
let tokens = ["ro", "rw", "rwa"]
.iter()
.map(|n| Token {
@@ -390,7 +400,8 @@ mod tests {
sha256: config::hash(&format!("secret-{n}")),
})
.collect();
router(Directory::new(Store::scratch().await, tokens).unwrap())
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) {
@@ -411,8 +422,8 @@ mod tests {
#[tokio::test]
async fn each_access_level_does_what_it_may_and_no_more() {
let app = app().await;
let snap = serde_json::json!([{ "url": "https://x.test/a.xml", "subscribers": 2 }]);
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);
@@ -424,9 +435,11 @@ mod tests {
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());
@@ -453,6 +466,7 @@ mod tests {
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();

79
crates/directory/src/store.rs
View File

@@ -79,6 +79,8 @@ const SCHEMA: &[&str] = &[
instance TEXT NOT NULL,
feed_id TEXT NOT NULL,
subscribers BIGINT NOT NULL,
category TEXT,
subcategory TEXT,
PRIMARY KEY (instance, feed_id)
)",
"CREATE INDEX IF NOT EXISTS snapshots_feed ON snapshots (feed_id)",
@@ -136,6 +138,11 @@ impl Store {
for sql in SCHEMA {
db.execute_unprepared(sql).await?;
}
// Added after the first deploy: a database made before has the table without them.
if db.execute_unprepared("SELECT category FROM snapshots WHERE false").await.is_err() {
db.execute_unprepared("ALTER TABLE snapshots ADD COLUMN category TEXT").await?;
db.execute_unprepared("ALTER TABLE snapshots ADD COLUMN subcategory TEXT").await?;
}
Ok(Self { db })
}
@@ -182,8 +189,12 @@ impl Store {
}
// The same feed twice in one snapshot counts once, at its larger number.
let n = e.subscribers.max(0);
let slot = keep.entry(id).or_insert((url, n));
let path = e.category.as_deref().map(str::trim).filter(|c| !c.is_empty()).map(ipx_feed::category_path);
let slot = keep.entry(id).or_insert((url, n, path.clone()));
slot.1 = slot.1.max(n);
if slot.2.is_none() {
slot.2 = path;
}
}
result.accepted = keep.len();
@@ -191,15 +202,21 @@ impl Store {
let backend = tx.get_database_backend();
let run = |sql: &str, values: Vec<Value>| Statement::from_sql_and_values(backend, sql, values);
tx.execute_raw(run("DELETE FROM snapshots WHERE instance = $1", vec![instance.into()])).await?;
for (id, (url, n)) in &keep {
for (id, (url, n, path)) in &keep {
tx.execute_raw(run(
"INSERT INTO feeds (id, url) VALUES ($1, $2) ON CONFLICT (id) DO NOTHING",
vec![id.clone().into(), url.clone().into()],
))
.await?;
tx.execute_raw(run(
"INSERT INTO snapshots (instance, feed_id, subscribers) VALUES ($1, $2, $3)",
vec![instance.into(), id.clone().into(), (*n).into()],
"INSERT INTO snapshots (instance, feed_id, subscribers, category, subcategory) VALUES ($1, $2, $3, $4, $5)",
vec![
instance.into(),
id.clone().into(),
(*n).into(),
path.as_ref().map(|p| p.0.clone()).into(),
path.as_ref().and_then(|p| p.1.clone()).into(),
],
))
.await?;
}
@@ -231,9 +248,14 @@ impl Store {
Ok(n)
}
const LISTING: &str = "SELECT f.id, f.url, f.title, f.image, f.category, f.subcategory, f.podcast,
CAST(sum(s.subscribers) AS BIGINT) AS subscribers
FROM feeds f JOIN snapshots s ON s.feed_id = f.id";
// The feed's own category, or else one a sending server's admin gave it.
// Only feeds read as a feed at least once (`items` set): never an OPML a server sent before it
// knew it was one, nor an address that has never worked.
const LISTING: &str = "SELECT f.id, f.url, f.title, f.image,
CASE WHEN f.category IS NULL THEN max(s.category) ELSE f.category END AS category,
CASE WHEN f.category IS NULL THEN max(s.subcategory) ELSE f.subcategory END AS subcategory,
f.podcast, CAST(sum(s.subscribers) AS BIGINT) AS subscribers
FROM feeds f JOIN snapshots s ON s.feed_id = f.id WHERE f.items IS NOT NULL";
fn listing(r: &sea_orm::QueryResult) -> Result<Listing> {
Ok(Listing {
@@ -268,7 +290,7 @@ impl Store {
/// One feed with its description and latest items.
pub async fn feed(&self, id: &str) -> Result<Option<FeedPage>> {
let sql = format!("{} WHERE f.id = $1 GROUP BY f.id", Self::LISTING);
let sql = format!("{} AND f.id = $1 GROUP BY f.id", Self::LISTING);
let Some(r) = self.rows(&sql, vec![id.into()]).await?.into_iter().next() else { return Ok(None) };
let listing = Self::listing(&r)?;
let r = self
@@ -419,11 +441,29 @@ fn failure_words(err: &str, failing_for: i64) -> String {
}
#[cfg(test)]
mod tests {
pub(crate) mod tests {
use super::*;
/// Reads the fixture feed as every feed waiting to be fetched, as the refresher would.
pub(crate) async fn fetch_all(s: &Store) {
for d in s.due(0, 100).await.unwrap() {
let parsed = ipx_feed::parse(include_bytes!("../../../tests/data/rss2.xml")).unwrap();
s.save(&d.id, Refreshed::Feed { parsed, etag: None, last_modified: None }).await.unwrap();
}
}
fn e(url: &str, n: i64) -> SnapshotEntry {
SnapshotEntry { url: url.into(), subscribers: n }
SnapshotEntry { url: url.into(), subscribers: n, category: None }
}
#[test]
fn a_failure_is_explained_after_a_day_and_never_names_its_address() {
let nothing = "the feed's address answered with nothing at all";
assert_eq!(failure_words(nothing, FLAG_AFTER_SECS), "This address answers with nothing; the site may be gone.");
// Not before it has been failing a day, as on the feed's own page in ipx (#143).
assert_eq!(failure_words(nothing, 3600), "Its last check failed.");
let raw = "error sending request for url (http://daily-quests.com/comic/?feed=atom): connection closed";
assert_eq!(failure_words(raw, FLAG_AFTER_SECS), "Its last check failed.");
}
#[test]
@@ -445,6 +485,9 @@ mod tests {
.unwrap();
assert_eq!(r.accepted, 2);
s.snapshot("b", &[e("https://x.test/a.xml", 2)]).await.unwrap();
// Nothing is listed until it has been read as a feed.
assert!(s.list().await.unwrap().is_empty());
fetch_all(&s).await;
let pop = s.popular(10).await.unwrap();
assert_eq!((pop[0].url.as_str(), pop[0].subscribers), ("https://x.test/a.xml", 7));
assert_eq!(pop[1].subscribers, 1);
@@ -480,6 +523,7 @@ mod tests {
.await
.unwrap();
assert_eq!((r.accepted, r.refused_private, r.refused_invalid), (1, 4, 1));
fetch_all(&s).await;
let id = s.list().await.unwrap()[0].id.clone();
assert!(s.block(&id).await.unwrap());
@@ -491,6 +535,21 @@ mod tests {
assert_eq!(s.snapshot("a", &[e("https://x.test/public.xml", 1)]).await.unwrap().accepted, 1);
}
#[tokio::test]
async fn a_feed_naming_no_category_takes_the_one_a_server_gave_it() {
let s = Store::scratch().await;
let entry = |c: Option<&str>| SnapshotEntry { url: "https://x.test/blog.xml".into(), subscribers: 1, category: c.map(Into::into) };
s.snapshot("a", &[entry(Some("Visual Arts"))]).await.unwrap();
let d = s.due(0, 10).await.unwrap().remove(0);
let mut parsed = ipx_feed::parse(include_bytes!("../../../tests/data/rss2.xml")).unwrap();
parsed.category = None;
s.save(&d.id, Refreshed::Feed { parsed, etag: None, last_modified: None }).await.unwrap();
let l = s.list().await.unwrap().remove(0);
assert_eq!((l.category.as_deref(), l.subcategory.as_deref()), (Some("Arts"), Some("Visual Arts")));
s.snapshot("a", &[entry(None)]).await.unwrap();
assert_eq!(s.list().await.unwrap()[0].category, None);
}
#[tokio::test]
async fn a_fetched_feed_shows_its_metadata_and_newest_items_but_no_files() {
let s = Store::scratch().await;