ipx-directory: the directory of public podcasts ipx servers share (#204)
A new crate and service, crates/directory, live at ipx-dir.sdf1.net. ipx servers send the public feeds their people subscribe to (POST /v1/snapshot, replacing what that token sent before); the directory fetches each feed's title, artwork, category, description and latest twenty items every six hours with ipx-feed, and lists them A to Z and by subscribers summed over every server. It holds metadata only, never a file or a file's address. Tokens are in its config file as SHA-256 hashes, each ro, rw or rwa, and the file is read again within a minute of a change, so adding and revoking one is an edit and nothing else; a revoked token takes its server's counts. At first they were to come from environment variables, which would have meant recreating the container to add one. A file that does not load keeps the tokens as they were. Every address sent is checked with ipx_feed::url_looks_private, the check ipx makes before sending, so one server's miss does not publish a paid feed to every other; an rwa token can remove and block what both miss. Deployed: ipx-directory in the ipx Arcane project, database ipx_directory (its own login) on the databases project's Postgres, which db-backup already covers, and a tunnel route with no Access in front. The store's tests run on Postgres as well as SQLite when IPX_TEST_DATABASE_URL is set. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
518
crates/directory/src/store.rs
Normal file
518
crates/directory/src/store.rs
Normal file
@@ -0,0 +1,518 @@
|
||||
//! 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, 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,
|
||||
PRIMARY KEY (instance, feed_id)
|
||||
)",
|
||||
"CREATE INDEX IF NOT EXISTS snapshots_feed ON snapshots (feed_id)",
|
||||
"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:").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")).await.unwrap()
|
||||
}
|
||||
|
||||
pub async fn open(url: &str) -> 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);
|
||||
}
|
||||
let db = sea_orm::Database::connect(opts).await?;
|
||||
for sql in SCHEMA {
|
||||
db.execute_unprepared(sql).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?)
|
||||
}
|
||||
|
||||
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 slot = keep.entry(id).or_insert((url, n));
|
||||
slot.1 = slot.1.max(n);
|
||||
}
|
||||
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)) 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()],
|
||||
))
|
||||
.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<_, _>>()?;
|
||||
for name in present.iter().filter(|n| !names.contains(n)) {
|
||||
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)
|
||||
}
|
||||
|
||||
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";
|
||||
|
||||
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!("{} WHERE 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 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)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn e(url: &str, n: i64) -> SnapshotEntry {
|
||||
SnapshotEntry { url: url.into(), subscribers: n }
|
||||
}
|
||||
|
||||
#[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();
|
||||
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);
|
||||
|
||||
// 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));
|
||||
|
||||
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_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");
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user