rays
cd7a9b8530
With apple_country set, the directory fetches Apple's top 100 podcasts and the top 50 of each of
its 19 categories for that country (itunes.apple.com/{cc}/rss/toppodcasts), looks up each
show's RSS address (/lookup, 100 ids a call, three seconds apart, as Apple asks for no more than
about twenty a minute), and stores them as the snapshot apple-podcasts with no subscribers. The
refresher fetches them like any feed; the most-subscribed chart is unchanged. Apple's primary
genre goes in as the snapshot's category, used only while a feed names none.
The same weekly run reads Apple's genre tree (ws/genres?id=26) and warns when it differs from
ipx_feed's CATEGORIES; on 2026-10-15 they were the same, all 19 and their subcategories.
A syncs table keeps when it last ran: at start if a week has passed, again weekly, an hour after
a failure. keep_instances spares apple-podcasts, which no token holds, and no token may take
the name. A trace a sync, a span each answer.
CLAUDE.md: builds one cargo command at a time, the load looked at before and after, after a
build chained with a test compile took Tower past 200.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
623 lines
28 KiB
Rust
623 lines
28 KiB
Rust
//! The directory's database: the feeds instances have sent, what each instance sent, and what
|
|
//! an admin has blocked. SQL written by hand, as ipx's is, so it runs on SQLite (tests) and
|
|
//! Postgres (production) both: `$1` parameters, `ON CONFLICT`, booleans as themselves.
|
|
|
|
use anyhow::Result;
|
|
use sea_orm::{ConnectionTrait, DatabaseConnection, Statement, TransactionTrait, Value};
|
|
|
|
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;
|
|
|
|
/// How long a feed's failure goes unexplained: a site down for an hour is not called gone (ipx#143).
|
|
const FLAG_AFTER_SECS: i64 = 24 * 3600;
|
|
|
|
pub struct Store {
|
|
db: DatabaseConnection,
|
|
}
|
|
|
|
/// What the refresher needs to ask for a feed again.
|
|
pub struct Due {
|
|
pub id: String,
|
|
pub url: String,
|
|
pub etag: Option<String>,
|
|
pub last_modified: Option<String>,
|
|
}
|
|
|
|
/// What a fetch found, to be stored.
|
|
pub enum Refreshed {
|
|
NotModified,
|
|
Feed {
|
|
parsed: ipx_feed::ParsedFeed,
|
|
etag: Option<String>,
|
|
last_modified: Option<String>,
|
|
},
|
|
Failed(String),
|
|
}
|
|
|
|
pub fn now() -> i64 {
|
|
std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map_or(0, |d| d.as_secs() as i64)
|
|
}
|
|
|
|
/// The address as every instance would write it: scheme and host lower-cased, a default port
|
|
/// and a fragment dropped. None for anything that is not http(s).
|
|
pub fn normalise(url: &str) -> Option<String> {
|
|
let mut u = url::Url::parse(url.trim()).ok()?;
|
|
if !matches!(u.scheme(), "http" | "https") || u.host_str().is_none() {
|
|
return None;
|
|
}
|
|
u.set_fragment(None);
|
|
Some(u.to_string())
|
|
}
|
|
|
|
/// A feed's id: the first 64 bits of its normalised address's SHA-256, so every instance, and
|
|
/// anyone else, arrives at the same id for the same feed without asking.
|
|
pub fn feed_id(normalised: &str) -> String {
|
|
use sha2::Digest;
|
|
sha2::Sha256::digest(normalised.as_bytes())[..8].iter().map(|b| format!("{b:02x}")).collect()
|
|
}
|
|
|
|
const SCHEMA: &[&str] = &[
|
|
"CREATE TABLE IF NOT EXISTS feeds (
|
|
id TEXT PRIMARY KEY,
|
|
url TEXT NOT NULL UNIQUE,
|
|
title TEXT,
|
|
image TEXT,
|
|
category TEXT,
|
|
subcategory TEXT,
|
|
description TEXT,
|
|
podcast BOOLEAN NOT NULL DEFAULT false,
|
|
items TEXT,
|
|
etag TEXT,
|
|
last_modified TEXT,
|
|
checked_at BIGINT,
|
|
last_error TEXT,
|
|
error_since BIGINT
|
|
)",
|
|
"CREATE TABLE IF NOT EXISTS snapshots (
|
|
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)",
|
|
// When each periodic job last finished: Apple's weekly sync (#252).
|
|
"CREATE TABLE IF NOT EXISTS syncs (
|
|
name TEXT PRIMARY KEY,
|
|
at BIGINT NOT NULL
|
|
)",
|
|
"CREATE TABLE IF NOT EXISTS blocked (
|
|
id TEXT PRIMARY KEY,
|
|
url TEXT,
|
|
title TEXT,
|
|
at BIGINT NOT NULL
|
|
)",
|
|
];
|
|
|
|
impl Store {
|
|
/// A database of its own for a test: a fresh schema on IPX_TEST_DATABASE_URL's Postgres when
|
|
/// that is set, as ipx's tests do, or else SQLite in memory.
|
|
#[cfg(test)]
|
|
pub async fn scratch() -> Self {
|
|
static N: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0);
|
|
let Some(url) = std::env::var("IPX_TEST_DATABASE_URL").ok().filter(|u| !u.is_empty()) else {
|
|
return Self::open("sqlite::memory:", false).await.unwrap();
|
|
};
|
|
let (pid, n) = (std::process::id(), N.fetch_add(1, std::sync::atomic::Ordering::Relaxed));
|
|
// A bare connection: Store::open would make the tables in the public schema, and the
|
|
// tests running at once raced each other doing it.
|
|
let admin = Self { db: sea_orm::Database::connect(url.as_str()).await.unwrap() };
|
|
// Schemas an earlier run left behind: anything of ours not made by this process.
|
|
let old = admin
|
|
.rows(&format!(r"SELECT nspname FROM pg_namespace WHERE nspname LIKE 'ipxd\_%' AND nspname NOT LIKE 'ipxd\_{pid}\_%'"), vec![])
|
|
.await
|
|
.unwrap();
|
|
for r in old {
|
|
let name: String = r.try_get_by_index(0).unwrap();
|
|
admin.db.execute_unprepared(&format!("DROP SCHEMA IF EXISTS {name} CASCADE")).await.unwrap();
|
|
}
|
|
let schema = format!("ipxd_{pid}_{n}");
|
|
admin.db.execute_unprepared(&format!("CREATE SCHEMA {schema}")).await.unwrap();
|
|
let sep = if url.contains('?') { '&' } else { '?' };
|
|
Self::open(&format!("{url}{sep}options=-c%20search_path%3D{schema}%20-c%20client_min_messages%3Dwarning"), false).await.unwrap()
|
|
}
|
|
|
|
/// `keep_open` for the server, which keeps its whole pool connected.
|
|
pub async fn open(url: &str, keep_open: bool) -> Result<Self> {
|
|
let mut url = url.to_owned();
|
|
if url.starts_with("postgres") && !url.contains("options=") {
|
|
// Postgres sends a notice for every CREATE ... IF NOT EXISTS on something that exists;
|
|
// asking for none keeps them out of the log, as ipx's connections do.
|
|
let sep = if url.contains('?') { '&' } else { '?' };
|
|
url = format!("{url}{sep}options=-c%20client_min_messages%3Dwarning");
|
|
}
|
|
let mut opts = sea_orm::ConnectOptions::new(url.clone());
|
|
opts.sqlx_logging(false);
|
|
// An in-memory SQLite database is one per connection: a pool of them would be several.
|
|
if url.contains(":memory:") {
|
|
opts.max_connections(1);
|
|
}
|
|
// The server keeps every connection open, as ipx's daemon does: a Postgres connection takes
|
|
// 80-330 ms to open (#241). Not a test, which would hold ten for one question.
|
|
if keep_open && url.starts_with("postgres") {
|
|
opts.max_connections(10).min_connections(10);
|
|
}
|
|
let db = sea_orm::Database::connect(opts).await?;
|
|
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 })
|
|
}
|
|
|
|
fn stmt(&self, sql: &str, values: Vec<Value>) -> Statement {
|
|
Statement::from_sql_and_values(self.db.get_database_backend(), sql, values)
|
|
}
|
|
|
|
async fn exec(&self, sql: &str, values: Vec<Value>) -> Result<u64> {
|
|
Ok(self.db.execute_raw(self.stmt(sql, values)).await?.rows_affected())
|
|
}
|
|
|
|
async fn rows(&self, sql: &str, values: Vec<Value>) -> Result<Vec<sea_orm::QueryResult>> {
|
|
Ok(self.db.query_all_raw(self.stmt(sql, values)).await?)
|
|
}
|
|
|
|
/// When the job `name` last finished, Unix seconds.
|
|
pub async fn synced_at(&self, name: &str) -> Result<Option<i64>> {
|
|
Ok(match self.rows("SELECT at FROM syncs WHERE name = $1", vec![name.into()]).await?.first() {
|
|
Some(r) => Some(r.try_get("", "at")?),
|
|
None => None,
|
|
})
|
|
}
|
|
|
|
pub async fn set_synced(&self, name: &str, at: i64) -> Result<()> {
|
|
self.exec(
|
|
"INSERT INTO syncs (name, at) VALUES ($1, $2) ON CONFLICT (name) DO UPDATE SET at = excluded.at",
|
|
vec![name.into(), at.into()],
|
|
)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
pub async fn ping(&self) -> Result<()> {
|
|
self.rows("SELECT 1", vec![]).await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Replaces everything `instance` sent before with `entries`. A private-looking address or a
|
|
/// blocked feed is left out and counted; the rest are added to the feeds, new ones to be
|
|
/// fetched by the refresher. Feeds no instance sends any more go.
|
|
pub async fn snapshot(&self, instance: &str, entries: &[SnapshotEntry]) -> Result<SnapshotResult> {
|
|
let blocked: std::collections::HashSet<String> =
|
|
self.rows("SELECT id FROM blocked", vec![]).await?.iter().map(|r| r.try_get("", "id")).collect::<Result<_, _>>()?;
|
|
let mut result = SnapshotResult::default();
|
|
let mut keep = std::collections::BTreeMap::new();
|
|
for e in entries {
|
|
let Some(url) = normalise(&e.url) else {
|
|
result.refused_invalid += 1;
|
|
continue;
|
|
};
|
|
// Checked here as well as by the sender: one instance's miss must not publish
|
|
// someone's paid feed to every other.
|
|
if ipx_feed::url_looks_private(&url) {
|
|
result.refused_private += 1;
|
|
continue;
|
|
}
|
|
let id = feed_id(&url);
|
|
if blocked.contains(&id) {
|
|
result.refused_blocked += 1;
|
|
continue;
|
|
}
|
|
// The same feed twice in one snapshot counts once, at its larger number.
|
|
let n = e.subscribers.max(0);
|
|
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();
|
|
|
|
let tx = self.db.begin().await?;
|
|
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, 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, 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?;
|
|
}
|
|
tx.execute_raw(run("DELETE FROM feeds WHERE id NOT IN (SELECT feed_id FROM snapshots)", vec![])).await?;
|
|
tx.commit().await?;
|
|
Ok(result)
|
|
}
|
|
|
|
/// Drops the snapshots of every instance not in `names`: a token taken out of the config
|
|
/// file takes its counts with it.
|
|
pub async fn keep_instances(&self, names: &[String]) -> Result<u64> {
|
|
let mut dropped = 0;
|
|
let present: Vec<String> = self
|
|
.rows("SELECT DISTINCT instance FROM snapshots", vec![])
|
|
.await?
|
|
.iter()
|
|
.map(|r| r.try_get("", "instance"))
|
|
.collect::<Result<_, _>>()?;
|
|
// Apple's charts are no token's, and stay until the next sync replaces them.
|
|
for name in present.iter().filter(|n| !names.contains(n) && n.as_str() != crate::apple::INSTANCE) {
|
|
dropped += self.drop_instance(name).await?;
|
|
}
|
|
Ok(dropped)
|
|
}
|
|
|
|
/// Drops what one instance sent; how many feeds it had.
|
|
pub async fn drop_instance(&self, name: &str) -> Result<u64> {
|
|
let n = self.exec("DELETE FROM snapshots WHERE instance = $1", vec![name.into()]).await?;
|
|
self.exec("DELETE FROM feeds WHERE id NOT IN (SELECT feed_id FROM snapshots)", vec![]).await?;
|
|
Ok(n)
|
|
}
|
|
|
|
// 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 {
|
|
id: r.try_get("", "id")?,
|
|
url: r.try_get("", "url")?,
|
|
title: r.try_get("", "title")?,
|
|
image: r.try_get("", "image")?,
|
|
category: r.try_get("", "category")?,
|
|
subcategory: r.try_get("", "subcategory")?,
|
|
podcast: r.try_get("", "podcast")?,
|
|
subscribers: r.try_get("", "subscribers")?,
|
|
})
|
|
}
|
|
|
|
/// Every feed, A to Z by title (one not fetched yet has none and sorts by its address).
|
|
///
|
|
/// ponytail: the whole list in one answer, as ipx's own Directory was; page it when it
|
|
/// grows past a few thousand feeds.
|
|
pub async fn list(&self) -> Result<Vec<Listing>> {
|
|
let sql = format!("{} GROUP BY f.id ORDER BY lower(coalesce(f.title, f.url)), f.id", Self::LISTING);
|
|
self.rows(&sql, vec![]).await?.iter().map(Self::listing).collect()
|
|
}
|
|
|
|
/// The most subscribed, most first.
|
|
pub async fn popular(&self, limit: u64) -> Result<Vec<Listing>> {
|
|
let sql = format!(
|
|
"{} GROUP BY f.id ORDER BY sum(s.subscribers) DESC, lower(coalesce(f.title, f.url)), f.id LIMIT $1",
|
|
Self::LISTING
|
|
);
|
|
self.rows(&sql, vec![(limit as i64).into()]).await?.iter().map(Self::listing).collect()
|
|
}
|
|
|
|
/// One feed with its description and latest items.
|
|
pub async fn feed(&self, id: &str) -> Result<Option<FeedPage>> {
|
|
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
|
|
.rows("SELECT description, items, last_error, error_since FROM feeds WHERE id = $1", vec![id.into()])
|
|
.await?
|
|
.into_iter()
|
|
.next()
|
|
.ok_or_else(|| anyhow::anyhow!("feed {id} went while it was read"))?;
|
|
let items: Option<String> = r.try_get("", "items")?;
|
|
let last_error: Option<String> = r.try_get("", "last_error")?;
|
|
let error_since: Option<i64> = r.try_get("", "error_since")?;
|
|
Ok(Some(FeedPage {
|
|
listing,
|
|
description: r.try_get("", "description")?,
|
|
failing: last_error.map(|e| failure_words(&e, now() - error_since.unwrap_or_else(now))),
|
|
items: items.and_then(|i| serde_json::from_str(&i).ok()).unwrap_or_default(),
|
|
}))
|
|
}
|
|
|
|
/// Removes a feed and blocks it, so no snapshot brings it back. False when it was neither
|
|
/// listed nor blocked already.
|
|
pub async fn block(&self, id: &str) -> Result<bool> {
|
|
let r = self.rows("SELECT url, title FROM feeds WHERE id = $1", vec![id.into()]).await?.into_iter().next();
|
|
let (url, title): (Option<String>, Option<String>) = match &r {
|
|
Some(r) => (r.try_get("", "url")?, r.try_get("", "title")?),
|
|
None => (None, None),
|
|
};
|
|
let added = self
|
|
.exec(
|
|
"INSERT INTO blocked (id, url, title, at) VALUES ($1, $2, $3, $4) ON CONFLICT (id) DO NOTHING",
|
|
vec![id.into(), url.into(), title.into(), now().into()],
|
|
)
|
|
.await?;
|
|
self.exec("DELETE FROM snapshots WHERE feed_id = $1", vec![id.into()]).await?;
|
|
let removed = self.exec("DELETE FROM feeds WHERE id = $1", vec![id.into()]).await?;
|
|
Ok(added + removed > 0)
|
|
}
|
|
|
|
/// Unblocks a feed; it comes back with the next snapshot that has it.
|
|
pub async fn unblock(&self, id: &str) -> Result<bool> {
|
|
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?
|
|
.iter()
|
|
.map(|r| Ok(Blocked { id: r.try_get("", "id")?, url: r.try_get("", "url")?, title: r.try_get("", "title")?, at: r.try_get("", "at")? }))
|
|
.collect()
|
|
}
|
|
|
|
/// Feeds never fetched, then those fetched longest ago, older than `every` seconds.
|
|
pub async fn due(&self, every: i64, limit: u64) -> Result<Vec<Due>> {
|
|
self.rows(
|
|
"SELECT id, url, etag, last_modified FROM feeds
|
|
WHERE checked_at IS NULL OR checked_at < $1
|
|
ORDER BY coalesce(checked_at, 0), id LIMIT $2",
|
|
vec![(now() - every).into(), (limit as i64).into()],
|
|
)
|
|
.await?
|
|
.iter()
|
|
.map(|r| {
|
|
Ok(Due {
|
|
id: r.try_get("", "id")?,
|
|
url: r.try_get("", "url")?,
|
|
etag: r.try_get("", "etag")?,
|
|
last_modified: r.try_get("", "last_modified")?,
|
|
})
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
pub async fn due_one(&self, id: &str) -> Result<Option<Due>> {
|
|
Ok(self
|
|
.rows("SELECT id, url, etag, last_modified FROM feeds WHERE id = $1", vec![id.into()])
|
|
.await?
|
|
.first()
|
|
.map(|r| -> Result<Due> {
|
|
Ok(Due {
|
|
id: r.try_get("", "id")?,
|
|
url: r.try_get("", "url")?,
|
|
etag: r.try_get("", "etag")?,
|
|
last_modified: r.try_get("", "last_modified")?,
|
|
})
|
|
})
|
|
.transpose()?)
|
|
}
|
|
|
|
pub async fn save(&self, id: &str, got: Refreshed) -> Result<()> {
|
|
let t = now();
|
|
match got {
|
|
Refreshed::NotModified => {
|
|
self.exec("UPDATE feeds SET checked_at = $1, last_error = NULL, error_since = NULL WHERE id = $2", vec![t.into(), id.into()]).await?;
|
|
}
|
|
Refreshed::Failed(err) => {
|
|
self.exec(
|
|
"UPDATE feeds SET checked_at = $1, last_error = $2, error_since = coalesce(error_since, $1) WHERE id = $3",
|
|
vec![t.into(), err.into(), id.into()],
|
|
)
|
|
.await?;
|
|
}
|
|
Refreshed::Feed { parsed, etag, last_modified } => {
|
|
let path = parsed.category.as_deref().map(ipx_feed::category_path);
|
|
let podcast = parsed.entries.iter().flat_map(|e| &e.enclosures).any(|e| {
|
|
e.mime.as_deref().is_some_and(|m| m.starts_with("audio/") || m.starts_with("video/"))
|
|
});
|
|
let mut entries = parsed.entries;
|
|
entries.sort_by_key(|e| std::cmp::Reverse(e.published));
|
|
// What an item is, never its files: their addresses are where a key would be.
|
|
let items: Vec<Item> = entries
|
|
.into_iter()
|
|
.take(ITEMS)
|
|
.map(|e| Item { guid: e.guid, title: e.title, link: e.link, published: e.published, duration: e.duration, description: e.description })
|
|
.collect();
|
|
self.exec(
|
|
"UPDATE feeds SET title = $1, image = $2, category = $3, subcategory = $4, description = $5,
|
|
podcast = $6, items = $7, etag = $8, last_modified = $9, checked_at = $10,
|
|
last_error = NULL, error_since = NULL
|
|
WHERE id = $11",
|
|
vec![
|
|
parsed.title.into(),
|
|
parsed.image.into(),
|
|
path.as_ref().map(|p| p.0.clone()).into(),
|
|
path.and_then(|p| p.1).into(),
|
|
parsed.description.into(),
|
|
podcast.into(),
|
|
serde_json::to_string(&items)?.into(),
|
|
etag.into(),
|
|
last_modified.into(),
|
|
t.into(),
|
|
id.into(),
|
|
],
|
|
)
|
|
.await?;
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
/// A feed's last error as its page says it: explained once it has been failing a day, or else
|
|
/// only that the check failed. Never the error itself, which is for the log.
|
|
fn failure_words(err: &str, failing_for: i64) -> String {
|
|
Some(err)
|
|
.filter(|_| failing_for >= FLAG_AFTER_SECS)
|
|
.and_then(ipx_feed::explain_failure)
|
|
.map_or_else(|| "Its last check failed.".to_owned(), |f| f.reason.to_owned())
|
|
}
|
|
|
|
#[cfg(test)]
|
|
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, 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]
|
|
fn every_instance_writes_one_feed_the_same_way() {
|
|
let a = normalise("HTTPS://Example.COM:443/feed.xml#top").unwrap();
|
|
assert_eq!(a, "https://example.com/feed.xml");
|
|
assert_eq!(feed_id(&a), feed_id(&normalise("https://example.com/feed.xml").unwrap()));
|
|
assert_eq!(feed_id(&a).len(), 16);
|
|
assert!(normalise("ftp://example.com/x").is_none());
|
|
assert!(normalise("not a url").is_none());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn snapshots_replace_what_an_instance_sent_and_add_up_across_instances() {
|
|
let s = Store::scratch().await;
|
|
let r = s
|
|
.snapshot("a", &[e("https://x.test/a.xml", 3), e("https://x.test/b.xml", 1), e("https://x.test/a.xml", 5)])
|
|
.await
|
|
.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);
|
|
|
|
// a no longer has b.xml: nobody does, so it goes.
|
|
s.snapshot("a", &[e("https://x.test/a.xml", 1)]).await.unwrap();
|
|
let all = s.list().await.unwrap();
|
|
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);
|
|
s.drop_instance("b").await.unwrap();
|
|
assert!(s.list().await.unwrap().is_empty());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn private_looking_addresses_and_blocked_feeds_are_refused() {
|
|
let s = Store::scratch().await;
|
|
let r = s
|
|
.snapshot(
|
|
"a",
|
|
&[
|
|
e("https://feeds.supercast.com/feeds/abc123", 1),
|
|
e("https://x.test/feed?token=secret", 1),
|
|
e("https://user:pass@x.test/feed", 1),
|
|
e("https://www.patreon.com/rss/show?auth=k", 1),
|
|
e("mailto:x@y", 1),
|
|
e("https://x.test/public.xml", 1),
|
|
],
|
|
)
|
|
.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());
|
|
assert!(s.list().await.unwrap().is_empty());
|
|
let r = s.snapshot("a", &[e("https://x.test/public.xml", 1)]).await.unwrap();
|
|
assert_eq!((r.accepted, r.refused_blocked), (0, 1));
|
|
assert_eq!(s.blocked().await.unwrap()[0].url.as_deref(), Some("https://x.test/public.xml"));
|
|
assert!(s.unblock(&id).await.unwrap());
|
|
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;
|
|
s.snapshot("a", &[e("https://x.test/feed.xml", 1)]).await.unwrap();
|
|
let due = s.due(3600, 10).await.unwrap();
|
|
assert_eq!(due.len(), 1);
|
|
let parsed = ipx_feed::parse(include_bytes!("../../../tests/data/rss2.xml")).unwrap();
|
|
let title = parsed.title.clone();
|
|
s.save(&due[0].id, Refreshed::Feed { parsed, etag: Some("e".into()), last_modified: None }).await.unwrap();
|
|
assert!(s.due(3600, 10).await.unwrap().is_empty());
|
|
let page = s.feed(&due[0].id).await.unwrap().unwrap();
|
|
assert_eq!(page.listing.title, title);
|
|
assert!(page.listing.podcast);
|
|
assert!(!page.items.is_empty() && page.items.len() <= ITEMS);
|
|
assert!(page.items.windows(2).all(|w| w[0].published >= w[1].published));
|
|
assert!(page.failing.is_none());
|
|
let json = serde_json::to_string(&page).unwrap();
|
|
assert!(!json.contains(".mp3"), "an item's file must not be in its page: {json}");
|
|
|
|
s.save(&due[0].id, Refreshed::Failed("HTTP 404 Not Found".into())).await.unwrap();
|
|
let page = s.feed(&due[0].id).await.unwrap().unwrap();
|
|
assert_eq!(page.failing.as_deref(), Some("Its last check failed."));
|
|
assert_eq!(page.listing.title, title, "a failure keeps what was found before");
|
|
}
|
|
}
|