From 611716d8b7ca2700187a43298020fbcc22a2d8bc Mon Sep 17 00:00:00 2001 From: rays Date: Fri, 18 Sep 2026 19:10:59 +0000 Subject: [PATCH] SeaORM: feeds and scanning; rusqlite gone The last nineteen functions move to SeaORM: recording feeds, items and enclosures, managed OPML feeds, folding WordPress's repeated files, and handing a Patreon creator's files to its shows. Two SQLite-only forms go: GLOB becomes a LIKE with the underscore escaped (broader, harmlessly: the fold still keys on `_=` and digits), and UPDATE OR IGNORE becomes an UPDATE ... WHERE NOT EXISTS. The two transactions are SeaORM transactions. With nothing left on it, rusqlite goes, with the SQL schema and migrate(). The entities are the schema: create_missing makes whatever tables and indexes a database lacks, from them, with CREATE ... IF NOT EXISTS. Production's schema already has every column migrate() added and none it dropped. Not SeaORM's schema sync, used until now: despite its docs it drops a unique index the entities do not describe, so it dropped users_name_lower on every open. Every `ipx` command then took a write lock, and against a daemon busy writing, `ipx status` -- the healthcheck -- failed 7 times in 15 where the old code failed none. Now 15 in 15, as before. On Postgres it would not have started. WAL is set only when a file is not already in it: setting it takes a lock that cannot wait out a busy daemon. Checked on copies of production: a forced scan of all 162 feeds against the real feeds with no database errors; the feed list, filters, sorts, search and the reaper's candidates against the old code on the same data, earlier in the branch. The column comments from the SQL schema move to the entities. Co-Authored-By: Claude Opus 5 --- CHANGELOG.md | 3 + Cargo.lock | 50 --- Cargo.toml | 3 +- src/db.rs | 964 ++++++++++++++++++----------------------------- src/entity.rs | 28 +- src/main.rs | 134 +++---- src/retention.rs | 8 +- src/web.rs | 28 +- 8 files changed, 485 insertions(+), 733 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index be53d5d..eb1b8da 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- The database is reached through SeaORM, on the way to Postgres (issue #18); it is still the + same SQLite file, and nothing you see changes. A database from before 0.7 has to be opened by a + 0.7 release first, which brings its tables up to date. - A pinned item sits at the top of its list, above everything else in whatever order you sort by, and moves there the moment you pin it. Sorting by the pin column itself still goes both ways, and Currently Listening keeps its own order. diff --git a/Cargo.lock b/Cargo.lock index ca082d4..244d03b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1129,18 +1129,6 @@ dependencies = [ "pin-project-lite", ] -[[package]] -name = "fallible-iterator" -version = "0.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2acce4a10f12dc2fb14a218589d4f1f62ef011b2d0cc4b3cb1bba8e94da14649" - -[[package]] -name = "fallible-streaming-iterator" -version = "0.1.9" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7360491ce676a36bf9bb3c56c1aa791658183a54d2744120f27285738d90465a" - [[package]] name = "fastrand" version = "2.5.0" @@ -1848,7 +1836,6 @@ dependencies = [ "quick-xml 0.42.0", "reqwest", "rss", - "rusqlite", "sea-orm", "serde", "serde_json", @@ -3289,16 +3276,6 @@ dependencies = [ "libc", ] -[[package]] -name = "rsqlite-vfs" -version = "0.1.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c51c9ae4df8a7fba42103df5c621fa3c37eccf3a3c650879e90fc48b11cc192c" -dependencies = [ - "hashbrown 0.16.1", - "thiserror 2.0.20", -] - [[package]] name = "rss" version = "2.1.1" @@ -3310,21 +3287,6 @@ dependencies = [ "quick-xml 0.41.0", ] -[[package]] -name = "rusqlite" -version = "0.39.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a0d2b0146dd9661bf67bb107c0bb2a55064d556eeb3fc314151b957f313bcd4e" -dependencies = [ - "bitflags 2.13.1", - "fallible-iterator", - "fallible-streaming-iterator", - "hashlink", - "libsqlite3-sys", - "smallvec", - "sqlite-wasm-rs", -] - [[package]] name = "rust_decimal" version = "1.43.0" @@ -3905,18 +3867,6 @@ dependencies = [ "lock_api", ] -[[package]] -name = "sqlite-wasm-rs" -version = "0.5.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dc3efc0da82635d7e1ced0053bbbfa8c7ab9645d0bf36ceb4f7127bb85315d75" -dependencies = [ - "cc", - "js-sys", - "rsqlite-vfs", - "wasm-bindgen", -] - [[package]] name = "sqlx" version = "0.9.0" diff --git a/Cargo.toml b/Cargo.toml index 847a9a0..cf598dc 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -18,8 +18,7 @@ percent-encoding = "2.3.2" quick-xml = { version = "0.42.0", features = ["escape-html"] } reqwest = { version = "0.13.5", default-features = false, features = ["rustls", "http2", "gzip", "stream", "json", "charset", "system-proxy"] } rss = "2.1.1" -rusqlite = { version = "0.39", features = ["bundled"] } -sea-orm = { version = "~2.0.3", default-features = false, features = ["sqlx-sqlite", "sqlx-postgres", "runtime-tokio-rustls", "macros", "with-json", "schema-sync", "sqlite-use-returning-for-3_35"] } +sea-orm = { version = "2.0.3", default-features = false, features = ["sqlx-sqlite", "sqlx-postgres", "runtime-tokio-rustls", "macros", "with-json", "sqlite-use-returning-for-3_35"] } serde = { version = "1.0.229", features = ["derive"] } serde_json = "1.0.151" tokio = { version = "1.53.1", features = ["rt-multi-thread", "macros", "fs", "io-util", "net", "sync", "time", "signal"] } diff --git a/src/db.rs b/src/db.rs index dee1bd1..aa9f478 100644 --- a/src/db.rs +++ b/src/db.rs @@ -1,23 +1,17 @@ -//! SQLite state. Replaces the per-feed .ipxd plists, history.dat and qmcache.dat. +//! The database, through SeaORM: SQLite today, Postgres to come (issue #18). Replaces the +//! per-feed .ipxd plists, history.dat and qmcache.dat. use anyhow::{Context, Result}; -use crate::entity::{enclosures, sessions, subscriptions, users}; -use rusqlite::{Connection, OptionalExtension, params}; +use crate::entity::{enclosures, entries, feeds, sessions, subscriptions, users}; use sea_orm::sea_query::{Expr, Func}; use sea_orm::{ ActiveModelTrait, ColumnTrait, ConnectionTrait, EntityTrait, PaginatorTrait, QueryFilter, QueryOrder, Set, Statement, }; use std::path::Path; -use std::sync::Mutex; -/// ponytail: one global connection mutex. Writes here are tiny and rare; move to a -/// spawn_blocking pool if a large feed count ever makes it contend. -/// -/// Mid-way through moving to SeaORM (issue #18): `orm` is the new connection, to the same -/// SQLite file, and functions move to it one at a time; `conn` goes when the last one has. +/// A pool of connections, where there was one connection behind a global lock. pub struct Db { - conn: Mutex, orm: sea_orm::DatabaseConnection, /// A test's database file, removed when the test is done with it. #[cfg(test)] @@ -35,26 +29,35 @@ impl Drop for Db { } } -/// Creates whatever the database is missing from the entities in `crate::entity`: tables, -/// columns, unique keys, foreign keys. It only ever adds. What an entity cannot say -- an -/// index on an expression, or on two columns -- follows as plain SQL both databases accept. +/// Creates whatever tables and indexes the database is missing, from the entities in +/// `crate::entity`, on SQLite or Postgres alike. It only ever creates: an existing table is left +/// as it is. What an entity cannot say -- an index on an expression, or on two columns -- +/// follows as plain SQL both databases accept. /// -/// ponytail: schema sync is experimental in SeaORM 2 and exempt from semver, hence the -/// `~2.0` pin in Cargo.toml; if it changes shape, move to sea-orm-migration files. -async fn sync(orm: &sea_orm::DatabaseConnection) -> Result<()> { +/// Not SeaORM's schema sync, which is experimental and, despite its docs, drops a unique index +/// the entities do not describe: it dropped users_name_lower on every open, so every `ipx` +/// command took a write lock, and the healthcheck's `ipx status` timed out behind a busy daemon. +/// On Postgres it would have failed outright, dropping that index as a constraint. +/// +/// ponytail: tables only, no columns. A column added to an existing table needs its own ALTER +/// here, as the old migrate() did for SQLite, or sea-orm-migration once there are several. A +/// database from before 0.7 took its last columns from that migrate(), so it upgrades through +/// a 0.7 release first. +async fn create_missing(orm: &sea_orm::DatabaseConnection) -> Result<()> { use crate::entity::*; - use sea_orm::ConnectionTrait; - orm.get_schema_builder() - .register(feeds::Entity) - .register(entries::Entity) - .register(enclosures::Entity) - .register(users::Entity) - .register(subscriptions::Entity) - .register(entry_state::Entity) - .register(sessions::Entity) - .sync(orm) - .await - .context("creating the schema")?; + use sea_orm::{ConnectionTrait, Schema}; + let schema = Schema::new(orm.get_database_backend()); + for mut table in [ + schema.create_table_from_entity(feeds::Entity), + schema.create_table_from_entity(entries::Entity), + schema.create_table_from_entity(enclosures::Entity), + schema.create_table_from_entity(users::Entity), + schema.create_table_from_entity(subscriptions::Entity), + schema.create_table_from_entity(entry_state::Entity), + schema.create_table_from_entity(sessions::Entity), + ] { + orm.execute(table.if_not_exists()).await.context("creating the schema")?; + } for sql in [ "CREATE INDEX IF NOT EXISTS enclosures_entry ON enclosures (feed_id, guid)", "CREATE UNIQUE INDEX IF NOT EXISTS users_name_lower ON users (lower(name))", @@ -64,6 +67,8 @@ async fn sync(orm: &sea_orm::DatabaseConnection) -> Result<()> { Ok(()) } +/// sqlx's defaults for SQLite are what ipx wants: foreign keys on, and a five-second wait for a +/// lock, which is what rusqlite was set to. async fn connect(path: &Path) -> Result { let mut opts = sea_orm::ConnectOptions::new(format!("sqlite://{}?mode=rwc", path.display())); opts.sqlx_logging(false); @@ -72,116 +77,6 @@ async fn connect(path: &Path) -> Result { .with_context(|| format!("opening {}", path.display())) } -const SCHEMA: &str = " -CREATE TABLE IF NOT EXISTS feeds ( - id TEXT PRIMARY KEY, - url TEXT NOT NULL, - title TEXT, - image TEXT, - -- The channel's first , for the Directory. - category TEXT, - etag TEXT, - last_modified TEXT, - last_checked INTEGER, - ttl_mins INTEGER, - last_error TEXT, - -- When the current run of failures began; NULL while the feed is healthy. Kept - -- through repeated failures so the UI can tell a blip (macmanx: failed once, fine an - -- hour later) from a feed that has been down for a day. - error_since INTEGER, - -- Came from a subscribed OPML that no longer lists it, but has downloads, so kept. - orphaned INTEGER NOT NULL DEFAULT 0, - -- The OPML subscription this feed came from. - group_id TEXT, - -- 1 = derived from an OPML and not written to config.toml. Writing 80-odd generated - -- entries into a hand-edited file made it unreadable; the OPML is the source of - -- truth, so they are re-derived instead. Customising one promotes it to config. - managed INTEGER NOT NULL DEFAULT 0 -); - -CREATE TABLE IF NOT EXISTS entries ( - feed_id TEXT NOT NULL, - guid TEXT NOT NULL, - title TEXT, - link TEXT, - published INTEGER, - description TEXT, - first_seen INTEGER NOT NULL, - image TEXT, - duration INTEGER, - episode INTEGER, - season INTEGER, - PRIMARY KEY (feed_id, guid) -); - --- url is UNIQUE: this is the dedupe key, and it subsumes the old history.dat pickle. --- A reaped file keeps its row with path = NULL and state = 'reaped', so a purged --- episode is never fetched a second time. -CREATE TABLE IF NOT EXISTS enclosures ( - id INTEGER PRIMARY KEY, - feed_id TEXT NOT NULL, - guid TEXT NOT NULL, - url TEXT NOT NULL UNIQUE, - mime TEXT, - length INTEGER, - path TEXT, - state TEXT NOT NULL, - bytes_done INTEGER NOT NULL DEFAULT 0, - downloaded_at INTEGER, - last_error TEXT -); - -CREATE INDEX IF NOT EXISTS enclosures_entry ON enclosures (feed_id, guid); - --- pass_hash is NULL for someone who only ever arrives through the proxy: there is no --- password to check, and leaving it empty is not the same as leaving it unset. -CREATE TABLE IF NOT EXISTS users ( - id INTEGER PRIMARY KEY, - name TEXT NOT NULL UNIQUE COLLATE NOCASE, - pass_hash TEXT, - is_admin INTEGER NOT NULL DEFAULT 0, - -- For whoever maintains the server. NULL where it is not known. - created INTEGER, - last_login INTEGER, - -- The theme chosen in Settings, and light, dark or auto. NULL until one is chosen. - theme TEXT, - theme_mode TEXT -); - --- What one person wants from a feed. The feed, its items and its files are shared; this --- is the part that is not. NULL in a column means: follow the feed's own setting. -CREATE TABLE IF NOT EXISTS subscriptions ( - user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, - feed_id TEXT NOT NULL, - keywords TEXT, - auto_download INTEGER, - allow_explicit INTEGER, - max_new_per_check INTEGER, - -- Pinned to the top of this person's feed list, a feed inside a folder included. - pinned INTEGER NOT NULL DEFAULT 0, - PRIMARY KEY (user_id, feed_id) -); - --- Read, kept and how far in. One row per person per item, created on first touch; --- an item nobody has touched has no row at all, which is what unread means. -CREATE TABLE IF NOT EXISTS entry_state ( - user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, - feed_id TEXT NOT NULL, - guid TEXT NOT NULL, - read INTEGER NOT NULL DEFAULT 0, - flagged INTEGER NOT NULL DEFAULT 0, - position INTEGER NOT NULL DEFAULT 0, - -- The length this person's player measured, beside the position it is measured against. - duration INTEGER, - PRIMARY KEY (user_id, feed_id, guid) -); - -CREATE TABLE IF NOT EXISTS sessions ( - token TEXT PRIMARY KEY, - user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, - seen INTEGER NOT NULL -); -"; /// One person's wants for one feed. `None` in a field means the feed's own setting stands. #[derive(Debug, Clone, Default)] @@ -261,62 +156,6 @@ pub struct Managed { pub group_id: String, } -/// Adds the columns later versions introduced and drops the ones they retired. CREATE TABLE IF -/// NOT EXISTS leaves a table that already exists alone, so an installed database needs both done -/// explicitly. Columns from before 0.3.0, the oldest version an upgrade may start from, need no -/// entry. -fn migrate(conn: &Connection) -> Result<()> { - let wanted: &[(&str, &str, &str)] = &[ - // For whoever maintains the server. An audit dropped `created` as unread on 2026-09-12, - // and it came back the same day with `last_login` beside it. - ("users", "created", "INTEGER"), - ("users", "last_login", "INTEGER"), - ("feeds", "error_since", "INTEGER"), - ("feeds", "category", "TEXT"), - ("entry_state", "duration", "INTEGER"), - // Kept per account so a theme follows you to another browser; it was in localStorage. - ("users", "theme", "TEXT"), - ("users", "theme_mode", "TEXT"), - ("subscriptions", "pinned", "INTEGER NOT NULL DEFAULT 0"), - ]; - let retired: &[(&str, &str)] = &[ - // Read state from before accounts, long since moved to entry_state. Two bugs came from - // queries still reading these after they stopped meaning anything. - ("entries", "read"), - ("entries", "flagged"), - ("entries", "position"), - // Written by every insert and read by nothing. - ("subscriptions", "created"), - ("sessions", "created"), - ]; - let has = |table: &str, column: &str| -> Result { - let names: Vec = conn - .prepare(&format!("PRAGMA table_info({table})"))? - .query_map([], |r| r.get(1))? - .collect::>()?; - Ok(names.iter().any(|c| c == column)) - }; - for (table, column, ty) in wanted { - if !has(table, column)? { - tracing::info!(table, column, "adding column"); - conn.execute_batch(&format!("ALTER TABLE {table} ADD COLUMN {column} {ty}"))?; - // A category is only read from a 200, and a feed with a validator mostly gets a - // 304, so most would never pick one up until the publisher changed something. - // Dropping the validators once makes each re-read on its normal schedule; leaving - // last_checked alone, unlike clear_validators, keeps them from all coming due at once. - if (*table, *column) == ("feeds", "category") { - conn.execute_batch("UPDATE feeds SET etag = NULL, last_modified = NULL")?; - } - } - } - for (table, column) in retired { - if has(table, column)? { - tracing::info!(table, column, "dropping column"); - conn.execute_batch(&format!("ALTER TABLE {table} DROP COLUMN {column}"))?; - } - } - Ok(()) -} /// What `ipx list` shows next to each configured feed. #[derive(Debug, Default)] @@ -341,17 +180,16 @@ impl Db { std::fs::create_dir_all(dir) .with_context(|| format!("creating {}", dir.display()))?; } - let conn = Connection::open(path) - .with_context(|| format!("opening {}", path.display()))?; - conn.pragma_update(None, "journal_mode", "WAL")?; - conn.pragma_update(None, "foreign_keys", "ON")?; - conn.pragma_update(None, "busy_timeout", 5000)?; - conn.execute_batch(SCHEMA).context("creating schema")?; - migrate(&conn).context("migrating schema")?; let orm = connect(path).await?; - sync(&orm).await?; + // WAL, so the healthcheck's `ipx status` reads while the daemon writes. It is a setting of + // the file, kept once made, and making it takes a lock that cannot wait out a busy + // daemon, so it is made only when the file is not already in WAL. + let mode = orm.query_one_raw(Statement::from_string(orm.get_database_backend(), "PRAGMA journal_mode")).await?; + if mode.and_then(|r| r.try_get_by_index::(0).ok()).as_deref() != Some("wal") { + orm.execute_unprepared("PRAGMA journal_mode = WAL").await.context("switching to WAL")?; + } + create_missing(&orm).await?; Ok(Self { - conn: Mutex::new(conn), orm, #[cfg(test)] tmp: None, @@ -359,8 +197,7 @@ impl Db { } /// A fresh database for a test, in a file of its own: two connections to one ":memory:" - /// would be two databases. Its schema comes from the entities alone (`sync`), not SCHEMA, - /// so every test also checks that the entities describe everything the queries need. + /// would be two databases. Its schema comes from the entities, as a real one's does. #[cfg(test)] pub async fn memory() -> Result { static N: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0); @@ -370,53 +207,46 @@ impl Db { N.fetch_add(1, std::sync::atomic::Ordering::Relaxed) )); let orm = connect(&path).await?; - sync(&orm).await?; - let conn = Connection::open(&path)?; - conn.pragma_update(None, "foreign_keys", "ON")?; - conn.pragma_update(None, "busy_timeout", 5000)?; - Ok(Self { conn: Mutex::new(conn), orm, tmp: Some(path) }) + create_missing(&orm).await?; + Ok(Self { orm, tmp: Some(path) }) } #[cfg(test)] - pub fn exec_for_test(&self, sql: &str) -> Result<()> { - self.conn.lock().unwrap().execute_batch(sql)?; + /// The first column of every row, for a test to check what a query left behind. + #[cfg(test)] + pub async fn i64s_for_test(&self, sql: &str) -> Vec { + self.rows(sql, vec![]).await.unwrap().iter().map(|r| r.try_get_by_index(0).unwrap()).collect() + } + + #[cfg(test)] + pub async fn strings_for_test(&self, sql: &str) -> Vec { + self.rows(sql, vec![]).await.unwrap().iter().map(|r| r.try_get_by_index(0).unwrap()).collect() + } + + #[cfg(test)] + pub async fn exec_for_test(&self, sql: &str) -> Result<()> { + self.orm.execute_unprepared(sql).await?; Ok(()) } - pub fn feed_summary(&self, feed_id: &str) -> Result { - let conn = self.conn.lock().unwrap(); - let mut sum: FeedSummary = conn - .query_row( - "SELECT title, image, last_checked, last_error, coalesce(orphaned, 0), error_since, - category - FROM feeds WHERE id = ?1", - [feed_id], - |r| { - Ok(FeedSummary { - title: r.get(0)?, - image: r.get(1)?, - last_checked: r.get(2)?, - last_error: r.get(3)?, - orphaned: r.get::<_, i64>(4)? != 0, - error_since: r.get(5)?, - category: r.get(6)?, - ..Default::default() - }) - }, - ) - .optional()? + pub async fn feed_summary(&self, feed_id: &str) -> Result { + let mut sum = feeds::Entity::find_by_id(feed_id.to_owned()) + .one(&self.orm) + .await? + .map(|f| FeedSummary { + title: f.title, + image: f.image, + last_checked: f.last_checked, + last_error: f.last_error, + orphaned: f.orphaned, + error_since: f.error_since, + category: f.category, + ..Default::default() + }) .unwrap_or_default(); - - sum.entries = conn.query_row( - "SELECT count(*) FROM entries WHERE feed_id = ?1", - [feed_id], - |r| r.get(0), - )?; - sum.downloaded = conn.query_row( - "SELECT count(*) FROM enclosures WHERE feed_id = ?1 AND path IS NOT NULL", - [feed_id], - |r| r.get(0), - )?; + sum.entries = + entries::Entity::find().filter(entries::Column::FeedId.eq(feed_id)).count(&self.orm).await? as i64; + sum.downloaded = self.downloaded_count(feed_id).await?; Ok(sum) } } @@ -432,27 +262,22 @@ pub struct HttpState { } impl Db { - pub fn http_state(&self, feed_id: &str) -> Result { - let conn = self.conn.lock().unwrap(); - Ok(conn - .query_row( - "SELECT etag, last_modified, last_checked, ttl_mins FROM feeds WHERE id = ?1", - [feed_id], - |r| { - Ok(HttpState { - etag: r.get(0)?, - last_modified: r.get(1)?, - last_checked: r.get(2)?, - ttl_mins: r.get::<_, Option>(3)?.map(|t| t.max(0) as u64), - }) - }, - ) - .optional()? + pub async fn http_state(&self, feed_id: &str) -> Result { + Ok(feeds::Entity::find_by_id(feed_id.to_owned()) + .one(&self.orm) + .await? + .map(|f| HttpState { + etag: f.etag, + last_modified: f.last_modified, + last_checked: f.last_checked, + ttl_mins: f.ttl_mins.map(|t| t.max(0) as u64), + }) .unwrap_or_default()) } /// Upsert after a successful poll. Clears any previous error. - pub fn record_feed( + #[allow(clippy::too_many_arguments)] + pub async fn record_feed( &self, feed_id: &str, url: &str, @@ -463,13 +288,13 @@ impl Db { image: Option<&str>, category: Option<&str>, ) -> Result<()> { - let conn = self.conn.lock().unwrap(); // category is taken as it comes, unlike title and image: a show that leaves a category // should leave the Directory's chip too. - conn.execute( + let o = |v: Option<&str>| sea_orm::Value::from(v.map(str::to_owned)); + self.exec( "INSERT INTO feeds (id, url, title, etag, last_modified, last_checked, ttl_mins, last_error, image, category) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, NULL, ?8, ?9) - ON CONFLICT(id) DO UPDATE SET + VALUES ($1, $2, $3, $4, $5, $6, $7, NULL, $8, $9) + ON CONFLICT (id) DO UPDATE SET url = excluded.url, title = coalesce(excluded.title, feeds.title), etag = excluded.etag, @@ -480,78 +305,106 @@ impl Db { category = excluded.category, last_error = NULL, error_since = NULL", - rusqlite::params![feed_id, url, title, etag, last_modified, now(), ttl_mins.map(|t| t as i64), image, category], - )?; + vec![ + feed_id.into(), + url.into(), + o(title), + o(etag), + o(last_modified), + now().into(), + ttl_mins.map(|t| t as i64).into(), + o(image), + o(category), + ], + ) + .await?; Ok(()) } /// 304, or any other poll that produced no new data: only the clock moves. - pub fn touch_feed(&self, feed_id: &str, url: &str) -> Result<()> { - let conn = self.conn.lock().unwrap(); - conn.execute( - "INSERT INTO feeds (id, url, last_checked) VALUES (?1, ?2, ?3) - ON CONFLICT(id) DO UPDATE SET last_checked = excluded.last_checked, + pub async fn touch_feed(&self, feed_id: &str, url: &str) -> Result<()> { + self.exec( + "INSERT INTO feeds (id, url, last_checked) VALUES ($1, $2, $3) + ON CONFLICT (id) DO UPDATE SET last_checked = excluded.last_checked, last_error = NULL, error_since = NULL", - rusqlite::params![feed_id, url, now()], - )?; + vec![feed_id.into(), url.into(), now().into()], + ) + .await?; Ok(()) } /// Forgets the cached ETag/Last-Modified. Those validators belong to the old URL, so /// keeping them across a URL change could produce a bogus 304 against the new one. - pub fn clear_validators(&self, feed_id: &str) -> Result<()> { - let conn = self.conn.lock().unwrap(); - conn.execute( - "UPDATE feeds SET etag = NULL, last_modified = NULL, last_checked = NULL WHERE id = ?1", - [feed_id], - )?; + pub async fn clear_validators(&self, feed_id: &str) -> Result<()> { + self.exec( + "UPDATE feeds SET etag = NULL, last_modified = NULL, last_checked = NULL WHERE id = $1", + vec![feed_id.into()], + ) + .await?; Ok(()) } - pub fn set_feed_error(&self, feed_id: &str, url: &str, msg: &str) -> Result<()> { - let conn = self.conn.lock().unwrap(); - let now = now(); - conn.execute( + pub async fn set_feed_error(&self, feed_id: &str, url: &str, msg: &str) -> Result<()> { + self.exec( "INSERT INTO feeds (id, url, last_checked, last_error, error_since) - VALUES (?1, ?2, ?3, ?4, ?3) - ON CONFLICT(id) DO UPDATE SET + VALUES ($1, $2, $3, $4, $3) + ON CONFLICT (id) DO UPDATE SET last_checked = excluded.last_checked, last_error = excluded.last_error, error_since = coalesce(feeds.error_since, excluded.error_since)", - rusqlite::params![feed_id, url, now, msg], - )?; + vec![feed_id.into(), url.into(), now().into(), msg.into()], + ) + .await?; Ok(()) } /// Returns true when this entry had not been seen before. /// - pub fn record_entry(&self, feed_id: &str, e: &crate::feed::Entry) -> Result { - let conn = self.conn.lock().unwrap(); - let inserted = conn.execute( - "INSERT OR IGNORE INTO entries - (feed_id, guid, title, link, published, description, first_seen, - image, duration, episode, season) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)", - rusqlite::params![ - feed_id, e.guid, e.title, e.link, e.published, e.description, now(), - e.image, e.duration, e.episode, e.season - ], - )?; - if inserted == 0 { - conn.execute( - "UPDATE entries SET - title = coalesce(?3, title), - description = coalesce(?4, description), - image = coalesce(?5, image), - duration = coalesce(?6, duration), - episode = coalesce(?7, episode), - season = coalesce(?8, season) - WHERE feed_id = ?1 AND guid = ?2", - rusqlite::params![ - feed_id, e.guid, e.title, e.description, - e.image, e.duration, e.episode, e.season + pub async fn record_entry(&self, feed_id: &str, e: &crate::feed::Entry) -> Result { + let inserted = self + .exec( + "INSERT INTO entries + (feed_id, guid, title, link, published, description, first_seen, + image, duration, episode, season) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) + ON CONFLICT DO NOTHING", + vec![ + feed_id.into(), + e.guid.clone().into(), + e.title.clone().into(), + e.link.clone().into(), + e.published.into(), + e.description.clone().into(), + now().into(), + e.image.clone().into(), + e.duration.into(), + e.episode.into(), + e.season.into(), ], - )?; + ) + .await?; + if inserted == 0 { + self.exec( + "UPDATE entries SET + title = coalesce($3, title), + description = coalesce($4, description), + image = coalesce($5, image), + duration = coalesce($6, duration), + episode = coalesce($7, episode), + season = coalesce($8, season) + WHERE feed_id = $1 AND guid = $2", + vec![ + feed_id.into(), + e.guid.clone().into(), + e.title.clone().into(), + e.description.clone().into(), + e.image.clone().into(), + e.duration.into(), + e.episode.into(), + e.season.into(), + ], + ) + .await?; } Ok(inserted == 1) } @@ -710,31 +563,33 @@ impl Db { /// Old entries that never had a file, or no longer have one. Enclosure rows stay -- /// they are the dedupe history. - pub fn prune_entries(&self, older_than: i64) -> Result { - let conn = self.conn.lock().unwrap(); - let n = conn.execute( - "DELETE FROM entries - WHERE coalesce(published, first_seen) < ?1 - AND NOT EXISTS ( - SELECT 1 FROM enclosures e - WHERE e.feed_id = entries.feed_id AND e.guid = entries.guid - AND e.path IS NOT NULL) - -- Starred by anyone keeps it, the same rule the reaper follows. - AND NOT EXISTS ( - SELECT 1 FROM entry_state s - WHERE s.feed_id = entries.feed_id AND s.guid = entries.guid - AND s.flagged = 1)", - [older_than], - )?; + pub async fn prune_entries(&self, older_than: i64) -> Result { + let n = self + .exec( + "DELETE FROM entries + WHERE coalesce(published, first_seen) < $1 + AND NOT EXISTS ( + SELECT 1 FROM enclosures e + WHERE e.feed_id = entries.feed_id AND e.guid = entries.guid + AND e.path IS NOT NULL) + -- Starred by anyone keeps it, the same rule the reaper follows. + AND NOT EXISTS ( + SELECT 1 FROM entry_state s + WHERE s.feed_id = entries.feed_id AND s.guid = entries.guid + AND s.flagged)", + vec![older_than.into()], + ) + .await?; // Whatever went takes everyone's read state with it, rather than leaving rows // pointing at an item that no longer exists. - conn.execute( + self.exec( "DELETE FROM entry_state WHERE NOT EXISTS ( SELECT 1 FROM entries e WHERE e.feed_id = entry_state.feed_id AND e.guid = entry_state.guid)", - [], - )?; - Ok(n) + vec![], + ) + .await?; + Ok(n as usize) } } @@ -1458,64 +1313,58 @@ impl Db { /// Names a feed without touching its conditional-GET validators. An OPML subscription /// takes its name from the document's own . - pub fn set_title(&self, feed_id: &str, title: &str) -> Result<()> { - let conn = self.conn.lock().unwrap(); - conn.execute( - "UPDATE feeds SET title = ?2 WHERE id = ?1 AND coalesce(title, '') != ?2", - rusqlite::params![feed_id, title], - )?; + pub async fn set_title(&self, feed_id: &str, title: &str) -> Result<()> { + self.exec( + "UPDATE feeds SET title = $2 WHERE id = $1 AND coalesce(title, '') <> $2", + vec![feed_id.into(), title.into()], + ) + .await?; Ok(()) } /// Records a feed that came from an OPML. Its settings are the parent's; only what /// identifies it is stored. - pub fn upsert_managed( - &self, - id: &str, - url: &str, - title: &str, - group_id: &str, - ) -> Result<()> { - let conn = self.conn.lock().unwrap(); - conn.execute( + pub async fn upsert_managed(&self, id: &str, url: &str, title: &str, group_id: &str) -> Result<()> { + self.exec( "INSERT INTO feeds (id, url, title, group_id, managed, orphaned) - VALUES (?1, ?2, ?3, ?4, 1, 0) - ON CONFLICT(id) DO UPDATE SET + VALUES ($1, $2, $3, $4, true, false) + ON CONFLICT (id) DO UPDATE SET url = excluded.url, title = coalesce(feeds.title, excluded.title), group_id = excluded.group_id, - managed = 1, - orphaned = 0", - rusqlite::params![id, url, title, group_id], - )?; + managed = true, + orphaned = false", + vec![id.into(), url.into(), title.into(), group_id.into()], + ) + .await?; Ok(()) } /// Every feed derived from an OPML, whichever group. - pub fn managed_feeds(&self) -> Result<Vec<Managed>> { - let conn = self.conn.lock().unwrap(); - let mut stmt = conn.prepare( + pub async fn managed_feeds(&self) -> Result<Vec<Managed>> { + self.rows( "SELECT id, url, title, group_id FROM feeds - WHERE managed = 1 AND group_id IS NOT NULL ORDER BY coalesce(title, id)", - )?; - Ok(stmt - .query_map([], |r| { - Ok(Managed { - id: r.get(0)?, - url: r.get(1)?, - title: r.get(2)?, - group_id: r.get(3)?, - }) - })? - .collect::<rusqlite::Result<Vec<_>>>()?) + WHERE managed AND group_id IS NOT NULL ORDER BY coalesce(title, id)", + vec![], + ) + .await? + .iter() + .map(|r| { + Ok(Managed { + id: r.try_get("", "id")?, + url: r.try_get("", "url")?, + title: r.try_get("", "title")?, + group_id: r.try_get("", "group_id")?, + }) + }) + .collect() } /// Forgets a derived feed entirely. Only for one with nothing downloaded. - pub fn drop_managed(&self, id: &str) -> Result<()> { - let conn = self.conn.lock().unwrap(); - conn.execute("DELETE FROM feeds WHERE id = ?1 AND managed = 1", [id])?; - conn.execute("DELETE FROM entries WHERE feed_id = ?1", [id])?; - conn.execute("DELETE FROM enclosures WHERE feed_id = ?1 AND path IS NULL", [id])?; + pub async fn drop_managed(&self, id: &str) -> Result<()> { + self.exec("DELETE FROM feeds WHERE id = $1 AND managed", vec![id.into()]).await?; + self.exec("DELETE FROM entries WHERE feed_id = $1", vec![id.into()]).await?; + self.exec("DELETE FROM enclosures WHERE feed_id = $1 AND path IS NULL", vec![id.into()]).await?; Ok(()) } @@ -1523,24 +1372,35 @@ impl Db { /// WordPress's numbered player URLs (`feed::same_file_key`). The first is the one the parser /// keeps, so it keeps its row, taking a repeat's file if it has none of its own; the repeats' /// rows go. Returns how many went and the copies left spare, for the caller to delete. - pub fn merge_repeated_enclosures(&self, key: impl Fn(&str) -> String) -> Result<(usize, Vec<String>)> { + pub async fn merge_repeated_enclosures(&self, key: impl Fn(&str) -> String) -> Result<(usize, Vec<String>)> { + use sea_orm::TransactionTrait; use std::collections::hash_map::Entry; - let mut conn = self.conn.lock().unwrap(); - let tx = conn.transaction()?; - let rows: Vec<(i64, String, String, String, Option<String>)> = { - let mut stmt = tx.prepare( - "SELECT id, feed_id, guid, url, path FROM enclosures + let backend = self.orm.get_database_backend(); + let tx = self.orm.begin().await?; + // Items with a URL carrying WordPress's `_=` parameter. A LIKE, with the underscore + // escaped, where SQLite had GLOB '*[?&]_=[0-9]*', which Postgres lacks. It lets through + // `_=` without a number too, which is harmless: `key` only folds `_=` and digits. + let rows = tx + .query_all_raw(Statement::from_string( + backend, + r"SELECT id, feed_id, guid, url, path FROM enclosures WHERE (feed_id, guid) IN - (SELECT feed_id, guid FROM enclosures WHERE url GLOB '*[?&]_=[0-9]*') + (SELECT feed_id, guid FROM enclosures + WHERE url LIKE '%?\_=%' ESCAPE '\' OR url LIKE '%&\_=%' ESCAPE '\') ORDER BY id", - )?; - stmt.query_map([], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?)))? - .collect::<rusqlite::Result<_>>()? - }; + )) + .await?; // The first row of each file, and whether it has the file on disk yet. let mut first: std::collections::HashMap<(String, String, String), (i64, bool)> = Default::default(); let (mut gone, mut spare) = (0, vec![]); - for (id, feed, guid, url, path) in rows { + for r in rows { + let (id, feed, guid, url, path): (i64, String, String, String, Option<String>) = ( + r.try_get("", "id")?, + r.try_get("", "feed_id")?, + r.try_get("", "guid")?, + r.try_get("", "url")?, + r.try_get("", "path")?, + ); match first.entry((feed, guid, key(&url))) { Entry::Vacant(v) => { v.insert((id, path.is_some())); @@ -1553,103 +1413,115 @@ impl Db { } else { // The only copy is the repeat's: Rands' episode 97 was downloaded // under its ?_=2 URL alone. - tx.execute( + tx.execute_raw(Statement::from_sql_and_values( + backend, "UPDATE enclosures SET (path, state, bytes_done, downloaded_at) = - (SELECT path, state, bytes_done, downloaded_at FROM enclosures WHERE id = ?2) - WHERE id = ?1", - params![*keep, id], - )?; + (SELECT path, state, bytes_done, downloaded_at FROM enclosures WHERE id = $2) + WHERE id = $1", + vec![(*keep).into(), id.into()], + )) + .await?; *has = true; } } - tx.execute("DELETE FROM enclosures WHERE id = ?1", [id])?; + tx.execute_raw(Statement::from_sql_and_values( + backend, + "DELETE FROM enclosures WHERE id = $1", + vec![id.into()], + )) + .await?; gone += 1; } } } - tx.commit()?; + tx.commit().await?; Ok((gone, spare)) } /// Stops treating a feed as derived, because it now has its own config entry. - pub fn unmanage(&self, id: &str) -> Result<()> { - let conn = self.conn.lock().unwrap(); - conn.execute("UPDATE feeds SET managed = 0 WHERE id = ?1", [id])?; + pub async fn unmanage(&self, id: &str) -> Result<()> { + self.exec("UPDATE feeds SET managed = false WHERE id = $1", vec![id.into()]).await?; Ok(()) } /// Empties a feed of its items, leaving its files alone. - pub fn clear_entries(&self, feed_id: &str) -> Result<()> { - let conn = self.conn.lock().unwrap(); - conn.execute("DELETE FROM entries WHERE feed_id = ?1", [feed_id])?; + pub async fn clear_entries(&self, feed_id: &str) -> Result<()> { + entries::Entity::delete_many().filter(entries::Column::FeedId.eq(feed_id)).exec(&self.orm).await?; Ok(()) } /// Every feed the database holds rows for, as (id, url), removed ones included. - pub fn feed_urls(&self) -> Result<Vec<(String, String)>> { - let conn = self.conn.lock().unwrap(); - let mut stmt = conn.prepare("SELECT id, coalesce(url, '') FROM feeds")?; - let out = stmt - .query_map([], |r| Ok((r.get(0)?, r.get(1)?)))? - .collect::<rusqlite::Result<_>>()?; - Ok(out) + pub async fn feed_urls(&self) -> Result<Vec<(String, String)>> { + Ok(feeds::Entity::find().all(&self.orm).await?.into_iter().map(|f| (f.id, f.url)).collect()) } /// Hands a feed in a group the enclosures its parent holds, as (guid, url), with everyone's /// read state for them. A Patreon creator read as one feed before it was split into shows /// owns every show's files, and `enclosures.url` is unique, so without this each show would /// list its items with nothing to play. - pub fn adopt(&self, parent: &str, child: &str, listed: &[(&str, &str)]) -> Result<()> { - let mut conn = self.conn.lock().unwrap(); - let holds: bool = conn.query_row( - "SELECT EXISTS (SELECT 1 FROM enclosures WHERE feed_id = ?1)", - [parent], - |r| r.get(0), - )?; + pub async fn adopt(&self, parent: &str, child: &str, listed: &[(&str, &str)]) -> Result<()> { + use sea_orm::TransactionTrait; + let holds = enclosures::Entity::find() + .filter(enclosures::Column::FeedId.eq(parent)) + .count(&self.orm) + .await? + > 0; if !holds { return Ok(()); // An OPML, or a creator already shared out. } - let tx = conn.transaction()?; + let backend = self.orm.get_database_backend(); + let tx = self.orm.begin().await?; for &(guid, url) in listed { - let moved = tx.execute( - "UPDATE enclosures SET feed_id = ?3, guid = ?4 WHERE url = ?1 AND feed_id = ?2", - params![url, parent, child, guid], - )?; + let moved = tx + .execute_raw(Statement::from_sql_and_values( + backend, + "UPDATE enclosures SET feed_id = $3, guid = $4 WHERE url = $1 AND feed_id = $2", + vec![url.into(), parent.into(), child.into(), guid.into()], + )) + .await? + .rows_affected(); if moved == 1 { - // Patreon gives a post the same guid in every feed it appears in. - tx.execute( - "UPDATE OR IGNORE entry_state SET feed_id = ?2 WHERE feed_id = ?1 AND guid = ?3", - params![parent, child, guid], - )?; + // Patreon gives a post the same guid in every feed it appears in. Someone who + // already has a row for it under the child keeps that one; SQLite's UPDATE OR + // IGNORE did this, and Postgres has no such thing. + tx.execute_raw(Statement::from_sql_and_values( + backend, + "UPDATE entry_state SET feed_id = $2 WHERE feed_id = $1 AND guid = $3 + AND NOT EXISTS (SELECT 1 FROM entry_state t + WHERE t.user_id = entry_state.user_id + AND t.feed_id = $2 AND t.guid = $3)", + vec![parent.into(), child.into(), guid.into()], + )) + .await?; } } - tx.commit()?; + tx.commit().await?; Ok(()) } /// A feed's enclosures skipped by one of its filters, by URL, with the reason: the verdicts a /// change of settings can overturn. A torrent held back while torrents are off is not a /// filter's call. - pub fn skipped_by_filter(&self, feed_id: &str) -> Result<std::collections::HashMap<String, String>> { - let conn = self.conn.lock().unwrap(); - let mut stmt = conn.prepare( + pub async fn skipped_by_filter(&self, feed_id: &str) -> Result<std::collections::HashMap<String, String>> { + self.rows( "SELECT url, last_error FROM enclosures - WHERE feed_id = ?1 AND state = 'skipped' AND last_error IS NOT NULL - AND last_error != 'torrents disabled'", - )?; - let out = stmt - .query_map([feed_id], |r| Ok((r.get(0)?, r.get(1)?)))? - .collect::<rusqlite::Result<_>>()?; - Ok(out) + WHERE feed_id = $1 AND state = 'skipped' AND last_error IS NOT NULL + AND last_error <> 'torrents disabled'", + vec![feed_id.into()], + ) + .await? + .iter() + .map(|r| Ok((r.try_get("", "url")?, r.try_get("", "last_error")?))) + .collect() } - pub fn set_orphaned(&self, feed_id: &str, on: bool) -> Result<()> { - let conn = self.conn.lock().unwrap(); - conn.execute( - "INSERT INTO feeds (id, url, orphaned) VALUES (?1, '', ?2) - ON CONFLICT(id) DO UPDATE SET orphaned = excluded.orphaned", - rusqlite::params![feed_id, on as i64], - )?; + pub async fn set_orphaned(&self, feed_id: &str, on: bool) -> Result<()> { + self.exec( + "INSERT INTO feeds (id, url, orphaned) VALUES ($1, '', $2) + ON CONFLICT (id) DO UPDATE SET orphaned = excluded.orphaned", + vec![feed_id.into(), on.into()], + ) + .await?; Ok(()) } @@ -1704,50 +1576,22 @@ mod tests { (5,'f','c','https://x/c.mp3?_=1','/d/c.mp3','done'), (6,'f','d','https://x/d1.mp3?_=1','/d/d1.mp3','done'), (7,'f','d','https://x/d2.mp3?_=2','/d/d2.mp3','done');", - ) + ).await .unwrap(); let key = crate::feed::same_file_key; - assert_eq!(db.merge_repeated_enclosures(key).unwrap(), (2, vec!["/d/a.mp3".to_string()])); - { - let conn = db.conn.lock().unwrap(); - let ids: Vec<i64> = conn - .prepare("SELECT id FROM enclosures ORDER BY id") - .unwrap() - .query_map([], |r| r.get(0)) - .unwrap() - .collect::<rusqlite::Result<_>>() - .unwrap(); - assert_eq!(ids, [1, 3, 5, 6, 7], "a lone ?_=1 and two different files stay"); - let (path, state): (String, String) = - conn.query_row("SELECT path, state FROM enclosures WHERE id = 3", [], |r| Ok((r.get(0)?, r.get(1)?))).unwrap(); - assert_eq!((path.as_str(), state.as_str()), ("/d/b.mp3", "done"), "the only copy moves, not deleted"); - } - assert_eq!(db.merge_repeated_enclosures(key).unwrap(), (0, vec![]), "and only once"); - } - - #[test] - fn adding_category_drops_validators_once_and_keeps_the_schedule() { - let conn = Connection::open_in_memory().unwrap(); - conn.execute_batch(SCHEMA).unwrap(); - // A database from before the column, holding a feed that would answer 304. - conn.execute_batch( - "ALTER TABLE feeds DROP COLUMN category; - INSERT INTO feeds (id, url, etag, last_modified, last_checked) VALUES ('f','u','e','lm',5);", - ) - .unwrap(); - let row = |conn: &Connection| -> (Option<String>, Option<String>, Option<i64>) { - conn.query_row("SELECT etag, last_modified, last_checked FROM feeds", [], |r| { - Ok((r.get(0)?, r.get(1)?, r.get(2)?)) - }) - .unwrap() - }; - migrate(&conn).unwrap(); - assert_eq!(row(&conn), (None, None, Some(5)), "re-read on its normal schedule"); - - // Only the once: every later open keeps the validators the next poll stored. - conn.execute_batch("UPDATE feeds SET etag = 'e2'").unwrap(); - migrate(&conn).unwrap(); - assert_eq!(row(&conn).0.as_deref(), Some("e2")); + assert_eq!(db.merge_repeated_enclosures(key).await.unwrap(), (2, vec!["/d/a.mp3".to_string()])); + assert_eq!( + db.i64s_for_test("SELECT id FROM enclosures ORDER BY id").await, + [1, 3, 5, 6, 7], + "a lone ?_=1 and two different files stay" + ); + assert_eq!( + db.strings_for_test("SELECT path FROM enclosures WHERE id = 3 UNION ALL SELECT state FROM enclosures WHERE id = 3") + .await, + ["/d/b.mp3", "done"], + "the only copy moves, not deleted" + ); + assert_eq!(db.merge_repeated_enclosures(key).await.unwrap(), (0, vec![]), "and only once"); } #[tokio::test] @@ -1762,7 +1606,7 @@ mod tests { INSERT INTO enclosures (id, feed_id, guid, url, mime, length, state) VALUES (1,'f','a','u1','audio/mpeg',300,'pending'),(2,'g','b','u2','image/png',10,'pending'), (3,'f','c','u3','video/mp4',2000,'pending');", - ) + ).await .unwrap(); let order = async |col: &str, dir: &str| -> Vec<String> { db.entries_in(1, None, Filter::All, None, 0, 50, &order_sql(col, dir, false)).await @@ -1804,7 +1648,7 @@ mod tests { INSERT INTO entries (feed_id, guid, first_seen) VALUES ('f','a',0); INSERT INTO enclosures (id, feed_id, guid, url, path, state) VALUES (1,'f','a','u1','/tmp/a','done');", - ) + ).await .unwrap(); // Nobody has touched it: both others still have it unplayed. @@ -1828,7 +1672,7 @@ mod tests { "INSERT INTO users (id, name, is_admin) VALUES (1,'ray',1),(2,'sam',0); INSERT INTO entries (feed_id, guid, title, first_seen) VALUES ('f','a','One',100),('f','b','Two',200);", - ) + ).await .unwrap(); assert_eq!(db.unread_count(1, "f").await.unwrap(), 2); @@ -1861,69 +1705,15 @@ mod tests { #[tokio::test] async fn schema_is_idempotent_and_summary_handles_unknown_feeds() { let db = Db::memory().await.unwrap(); - // Re-running the schema must not fail: open() does this on every start. - db.conn.lock().unwrap().execute_batch(SCHEMA).unwrap(); + // Creating what is missing again must not fail: open() does it on every start. + create_missing(&db.orm).await.unwrap(); - let sum = db.feed_summary("never-seen").unwrap(); + let sum = db.feed_summary("never-seen").await.unwrap(); assert_eq!(sum.last_checked, None); assert_eq!(sum.entries, 0); assert_eq!(sum.downloaded, 0); } - #[test] - fn an_old_database_loses_its_retired_columns() { - let conn = Connection::open_in_memory().unwrap(); - // As open() has it: a DROP COLUMN on a table that references another is the part worth - // proving, and it has to work with the foreign keys switched on. - conn.pragma_update(None, "foreign_keys", "ON").unwrap(); - conn.execute_batch( - "CREATE TABLE entries (feed_id TEXT NOT NULL, guid TEXT NOT NULL, - first_seen INTEGER NOT NULL, read INTEGER NOT NULL DEFAULT 0, - flagged INTEGER NOT NULL DEFAULT 0, position INTEGER NOT NULL DEFAULT 0, - PRIMARY KEY (feed_id, guid)); - CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT NOT NULL UNIQUE COLLATE NOCASE, - pass_hash TEXT, is_admin INTEGER NOT NULL DEFAULT 0, created INTEGER NOT NULL); - CREATE TABLE subscriptions ( - user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, - feed_id TEXT NOT NULL, created INTEGER NOT NULL, PRIMARY KEY (user_id, feed_id)); - CREATE TABLE sessions (token TEXT PRIMARY KEY, - user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, - created INTEGER NOT NULL, seen INTEGER NOT NULL); - INSERT INTO users VALUES (1, 'ray', NULL, 1, 0); - INSERT INTO subscriptions VALUES (1, 'f', 0); - INSERT INTO sessions VALUES ('t', 1, 0, 0);", - ) - .unwrap(); - // The same order as open(): the schema leaves the old tables alone, migrate() fixes them. - conn.execute_batch(SCHEMA).unwrap(); - migrate(&conn).unwrap(); - let cols = |table: &str| -> Vec<String> { - conn.prepare(&format!("PRAGMA table_info({table})")) - .unwrap() - .query_map([], |r| r.get(1)) - .unwrap() - .collect::<rusqlite::Result<_>>() - .unwrap() - }; - for (table, gone) in [ - ("entries", &["read", "flagged", "position"][..]), - ("subscriptions", &["created"][..]), - ("sessions", &["created"][..]), - ] { - let cols = cols(table); - assert!(!cols.iter().any(|c| gone.contains(&c.as_str())), "{table}: {cols:?}"); - } - // users.created is not retired: it keeps what it held, and last_login joins it. - let users = cols("users"); - assert!(users.iter().any(|c| c == "last_login"), "{users:?}"); - assert_eq!(conn.query_row("SELECT created FROM users", [], |r| r.get::<_, i64>(0)).unwrap(), 0); - // And the rows come through it. - let kept: i64 = conn - .query_row("SELECT count(*) FROM subscriptions JOIN sessions USING (user_id)", [], |r| r.get(0)) - .unwrap(); - assert_eq!(kept, 1); - } - #[tokio::test] async fn a_renamed_account_keeps_everything_but_its_name() { let db = Db::memory().await.unwrap(); @@ -1948,10 +1738,10 @@ mod tests { db.signed_in(id).await.unwrap(); let first = get().await.last_login.unwrap(); // Within the hour, the proxy vouching again writes nothing; after it, it does. - db.exec_for_test(&format!("UPDATE users SET last_login = {} WHERE id = {id}", first - 60)).unwrap(); + db.exec_for_test(&format!("UPDATE users SET last_login = {} WHERE id = {id}", first - 60)).await.unwrap(); db.signed_in(id).await.unwrap(); assert_eq!(get().await.last_login, Some(first - 60)); - db.exec_for_test(&format!("UPDATE users SET last_login = {} WHERE id = {id}", first - 7200)).unwrap(); + db.exec_for_test(&format!("UPDATE users SET last_login = {} WHERE id = {id}", first - 7200)).await.unwrap(); db.signed_in(id).await.unwrap(); assert!(get().await.last_login.unwrap() >= first); } @@ -1961,18 +1751,16 @@ mod tests { // Cutting this along with the dead read columns left the browser suite's admin with an // empty sidebar: it is how a fresh install's first account gets config.toml's feeds. let db = Db::memory().await.unwrap(); - db.exec_for_test("INSERT INTO users (id, name, is_admin) VALUES (1,'admin',1);") + db.exec_for_test("INSERT INTO users (id, name, is_admin) VALUES (1,'admin',1);").await .unwrap(); - let subs = || -> i64 { - db.conn.lock().unwrap().query_row("SELECT count(*) FROM subscriptions", [], |r| r.get(0)).unwrap() - }; + let subs = async || db.i64s_for_test("SELECT count(*) FROM subscriptions").await[0]; let catalogue = ["a".to_string(), "b".to_string()]; assert_eq!(db.adopt_catalogue(1, &catalogue).await.unwrap(), 2); - assert_eq!(subs(), 2); + assert_eq!(subs().await, 2); // Once anyone subscribes to anything it never runs again, so an unsubscribe sticks. db.unsubscribe(1, "a").await.unwrap(); assert_eq!(db.adopt_catalogue(1, &catalogue).await.unwrap(), 0); - assert_eq!(subs(), 1); + assert_eq!(subs().await, 1); } #[tokio::test] @@ -2006,7 +1794,7 @@ mod tests { -- Opened, and so read, but 42 of 900 seconds in: still Currently Listening. -- Filtering on read hid exactly these (issue #14). (7,'f','h',1,0,42);", - ) + ).await .unwrap(); for f in [Filter::All, Filter::Unread, Filter::Downloaded, Filter::Flagged, Filter::InProgress] { @@ -2061,7 +1849,7 @@ mod tests { (1,'f','old','u-old','pending'), (2,'f','mid','u-mid','pending'), (3,'f','new','u-new','pending');", - ) + ).await .unwrap(); let got: Vec<String> = db.pending("f", 2).await.unwrap().into_iter().map(|p| p.url).collect(); assert_eq!(got, vec!["u-new", "u-mid"], "newest first, oldest left for later"); @@ -2076,16 +1864,13 @@ mod tests { (2,'f','b','u2','pending',NULL), (3,'f','c','u3','downloading','/tmp/already-here'), (4,'f','d','u4','done','/tmp/x');", - ) + ).await .unwrap(); assert_eq!(db.requeue_interrupted().await.unwrap(), 1, "only the in-flight, fileless one"); - let conn = db.conn.lock().unwrap(); - let state = |id: i64| -> String { - conn.query_row("SELECT state FROM enclosures WHERE id = ?1", [id], |r| r.get(0)).unwrap() - }; - assert_eq!(state(1), "pending"); - assert_eq!(state(3), "downloading", "it has a file; leave it alone"); - assert_eq!(state(4), "done"); + let state = async |id: i64| db.strings_for_test(&format!("SELECT state FROM enclosures WHERE id = {id}")).await; + assert_eq!(state(1).await, ["pending"]); + assert_eq!(state(3).await, ["downloading"], "it has a file; leave it alone"); + assert_eq!(state(4).await, ["done"]); } #[tokio::test] @@ -2099,29 +1884,25 @@ mod tests { (3,'creator','c','u3','skipped',NULL,'explicit'), (4,'other','d','u4','skipped',NULL,'torrents disabled'); INSERT INTO entry_state (user_id, feed_id, guid, read) VALUES (1,'creator','a',1);", - ) + ).await .unwrap(); - db.adopt("creator", "show", &[("a", "u1"), ("b", "u2"), ("d", "u4")]).unwrap(); - { - let conn = db.conn.lock().unwrap(); - let owner = |id: i64| -> String { - conn.query_row("SELECT feed_id FROM enclosures WHERE id = ?1", [id], |r| r.get(0)).unwrap() - }; - assert_eq!(owner(1), "show", "a downloaded file moves with its item"); - assert_eq!(owner(2), "show"); - assert_eq!(owner(3), "creator", "this show does not list it"); - assert_eq!(owner(4), "other", "only the parent's are taken"); - let read: String = conn - .query_row("SELECT feed_id FROM entry_state WHERE user_id = 1 AND guid = 'a'", [], |r| r.get(0)) - .unwrap(); - assert_eq!(read, "show", "what you had read stays read"); - } + db.adopt("creator", "show", &[("a", "u1"), ("b", "u2"), ("d", "u4")]).await.unwrap(); + let owner = async |id: i64| db.strings_for_test(&format!("SELECT feed_id FROM enclosures WHERE id = {id}")).await; + assert_eq!(owner(1).await, ["show"], "a downloaded file moves with its item"); + assert_eq!(owner(2).await, ["show"]); + assert_eq!(owner(3).await, ["creator"], "this show does not list it"); + assert_eq!(owner(4).await, ["other"], "only the parent's are taken"); + assert_eq!( + db.strings_for_test("SELECT feed_id FROM entry_state WHERE user_id = 1 AND guid = 'a'").await, + ["show"], + "what you had read stays read" + ); // Only a filter's verdict can be overturned by a change of settings. - let skipped = db.skipped_by_filter("show").unwrap(); + let skipped = db.skipped_by_filter("show").await.unwrap(); assert_eq!(skipped.get("u2").map(String::as_str), Some("explicit")); assert_eq!(skipped.len(), 1); - assert!(db.skipped_by_filter("other").unwrap().is_empty(), "torrents disabled is not a filter"); + assert!(db.skipped_by_filter("other").await.unwrap().is_empty(), "torrents disabled is not a filter"); } #[tokio::test] @@ -2131,7 +1912,7 @@ mod tests { "INSERT INTO users (id, name, is_admin) VALUES (1,'ray',1),(2,'sam',0); INSERT INTO subscriptions (user_id, feed_id, allow_explicit) VALUES (1,'group',1),(1,'show',NULL),(2,'group',1),(2,'show',0);", - ) + ).await .unwrap(); let explicit = async |group| -> Vec<Option<bool>> { let mut v: Vec<_> = @@ -2146,19 +1927,19 @@ mod tests { #[tokio::test] async fn error_since_marks_the_start_of_a_run_of_failures_and_clears_on_success() { let db = Db::memory().await.unwrap(); - db.set_feed_error("f", "http://x", "HTTP 404").unwrap(); + db.set_feed_error("f", "http://x", "HTTP 404").await.unwrap(); // Backdate it, as if this feed had already been failing a while, so a second // failure landing "now" is distinguishable from the first. - db.exec_for_test("UPDATE feeds SET error_since = error_since - 3600 WHERE id = 'f'").unwrap(); - let first = db.feed_summary("f").unwrap().error_since.unwrap(); + db.exec_for_test("UPDATE feeds SET error_since = error_since - 3600 WHERE id = 'f'").await.unwrap(); + let first = db.feed_summary("f").await.unwrap().error_since.unwrap(); // macmanx: failed once, read fine an hour later. A second failure must not push // error_since forward -- the UI decides "failing for a day" from the first one. - db.set_feed_error("f", "http://x", "HTTP 404").unwrap(); - assert_eq!(db.feed_summary("f").unwrap().error_since, Some(first)); + db.set_feed_error("f", "http://x", "HTTP 404").await.unwrap(); + assert_eq!(db.feed_summary("f").await.unwrap().error_since, Some(first)); - db.touch_feed("f", "http://x").unwrap(); - let after = db.feed_summary("f").unwrap(); + db.touch_feed("f", "http://x").await.unwrap(); + let after = db.feed_summary("f").await.unwrap(); assert_eq!(after.last_error, None); assert_eq!(after.error_since, None, "a clean check ends the run of failures"); } @@ -2181,9 +1962,8 @@ mod tests { #[tokio::test] async fn enclosure_url_is_the_dedupe_key() { let db = Db::memory().await.unwrap(); - let conn = db.conn.lock().unwrap(); let insert = "INSERT INTO enclosures (feed_id, guid, url, state) VALUES ('f', 'g', 'http://x/a.mp3', 'pending')"; - conn.execute(insert, []).unwrap(); - assert!(conn.execute(insert, []).is_err(), "duplicate url must be rejected"); + db.exec_for_test(insert).await.unwrap(); + assert!(db.exec_for_test(insert).await.is_err(), "duplicate url must be rejected"); } } diff --git a/src/entity.rs b/src/entity.rs index f8a05cb..a8ff7bb 100644 --- a/src/entity.rs +++ b/src/entity.rs @@ -1,6 +1,6 @@ //! The database's tables as SeaORM entities: the one description of the schema, from which -//! `Db::open` creates what a database is missing, on SQLite or Postgres alike (see `db::sync`). -//! What each column means is in the comments on `db::SCHEMA`, the SQLite schema these match. +//! `Db::open` creates what a database is missing, on SQLite or Postgres alike (see +//! `db::create_missing`). Times are Unix seconds. pub mod feeds { use sea_orm::entity::prelude::*; @@ -16,6 +16,7 @@ pub mod feeds { pub title: Option<String>, #[sea_orm(column_type = "Text", nullable)] pub image: Option<String>, + /// The channel's first <itunes:category>, for the Directory. #[sea_orm(column_type = "Text", nullable)] pub category: Option<String>, #[sea_orm(column_type = "Text", nullable)] @@ -26,11 +27,19 @@ pub mod feeds { pub ttl_mins: Option<i64>, #[sea_orm(column_type = "Text", nullable)] pub last_error: Option<String>, + /// When the current run of failures began; NULL while the feed is healthy. Kept through + /// repeated failures so the UI can tell a blip (macmanx: failed once, fine an hour + /// later) from a feed that has been down for a day. pub error_since: Option<i64>, + /// Came from a subscribed OPML that no longer lists it, but has downloads, so kept. #[sea_orm(default_value = false)] pub orphaned: bool, + /// The OPML subscription this feed came from. #[sea_orm(column_type = "Text", nullable)] pub group_id: Option<String>, + /// Derived from an OPML and not written to config.toml. Writing 80-odd generated entries + /// into a hand-edited file made it unreadable; the OPML is the source of truth, so they + /// are re-derived instead. Customising one promotes it to config. #[sea_orm(default_value = false)] pub managed: bool, } @@ -84,7 +93,8 @@ pub mod enclosures { pub feed_id: String, #[sea_orm(column_type = "Text")] pub guid: String, - /// The dedupe key: one file serves every subscriber. + /// The dedupe key, and the reason one file serves every subscriber. A reaped file keeps + /// its row with path NULL and state 'reaped', so a purged episode is never fetched again. #[sea_orm(unique, column_type = "Text")] pub url: String, #[sea_orm(column_type = "Text", nullable)] @@ -115,16 +125,20 @@ pub mod users { pub struct Model { #[sea_orm(primary_key)] pub id: i64, - /// Unique without regard to case: `db::sync` adds the index on lower(name), which + /// Unique without regard to case: `db::create_missing` adds the index on lower(name), which /// works the same on both databases where SQLite's COLLATE NOCASE does not. #[sea_orm(column_type = "Text")] pub name: String, + /// NULL for someone who only ever arrives through the proxy: there is no password to + /// check, and leaving it empty is not the same as leaving it unset. #[sea_orm(column_type = "Text", nullable)] pub pass_hash: Option<String>, #[sea_orm(default_value = false)] pub is_admin: bool, + /// For whoever maintains the server. NULL where it is not known. pub created: Option<i64>, pub last_login: Option<i64>, + /// The theme chosen in Settings, and light, dark or auto. NULL until one is chosen. #[sea_orm(column_type = "Text", nullable)] pub theme: Option<String>, #[sea_orm(column_type = "Text", nullable)] @@ -161,6 +175,8 @@ macro_rules! owned_by_user { }; } +/// What one person wants from a feed. The feed, its items and its files are shared; this is the +/// part that is not. NULL in a column means: follow the feed's own setting. pub mod subscriptions { use sea_orm::entity::prelude::*; @@ -177,6 +193,7 @@ pub mod subscriptions { pub auto_download: Option<bool>, pub allow_explicit: Option<bool>, pub max_new_per_check: Option<i64>, + /// Pinned to the top of this person's feed list, a feed inside a folder included. #[sea_orm(default_value = false)] pub pinned: bool, } @@ -184,6 +201,8 @@ pub mod subscriptions { owned_by_user!(); } +/// Read, kept and how far in. One row per person per item, created on first touch; an item +/// nobody has touched has no row at all, which is what unread means. pub mod entry_state { use sea_orm::entity::prelude::*; @@ -202,6 +221,7 @@ pub mod entry_state { pub flagged: bool, #[sea_orm(default_value = 0)] pub position: i64, + /// The length this person's player measured, beside the position it is measured against. pub duration: Option<i64>, } diff --git a/src/main.rs b/src/main.rs index 6a227f0..595c3a4 100644 --- a/src/main.rs +++ b/src/main.rs @@ -220,7 +220,7 @@ async fn main() -> Result<()> { }); match cli.command { - Command::List => list(&ctx, &config_path), + Command::List => list(&ctx, &config_path).await, Command::Daemon { web } => daemon(ctx, config_path, web, events).await, Command::Add { url, folder, keywords } => { add(&ctx, &config_path, &url, folder, keywords).await @@ -228,7 +228,7 @@ async fn main() -> Result<()> { Command::Rm { feed } => rm(&ctx, &config_path, &feed).await, Command::User { cmd } => user_cmd(&ctx, cmd).await, Command::Import { file } => import(&ctx, &config_path, &file).await, - Command::Export { file } => export(&ctx, &file), + Command::Export { file } => export(&ctx, &file).await, _ => run(&ctx, wire_cmd.expect("only List and Daemon have no wire form")).await, } } @@ -351,7 +351,7 @@ async fn run(ctx: &Arc<Ctx>, cmd: Cmd) -> Result<()> { async fn status(ctx: &Ctx) -> Event { match ctx.db.counts().await { Ok((pending, downloaded)) => { - let feeds = subscriptions(ctx).map(|s| s.len()).unwrap_or(0); + let feeds = subscriptions(ctx).await.map(|s| s.len()).unwrap_or(0); Event::Status { feeds, pending, downloaded } } Err(e) => Event::Error { msg: format!("{e:#}") }, @@ -401,7 +401,7 @@ async fn daemon( // Before the parser knew WordPress's numbered player URLs, a file it listed twice was // downloaded twice. The repeats fold into the first, and their spare copies are deleted. - match ctx.db.merge_repeated_enclosures(feed::same_file_key) { + match ctx.db.merge_repeated_enclosures(feed::same_file_key).await { Ok((0, _)) => {} Ok((n, spare)) => { for path in &spare { @@ -431,7 +431,7 @@ async fn daemon( let mut ticker = tokio::time::interval(std::time::Duration::from_secs(60)); ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); tracing::info!( - feeds = subscriptions(&ctx).map(|s| s.len()).unwrap_or(0), + feeds = subscriptions(&ctx).await.map(|s| s.len()).unwrap_or(0), "daemon started" ); @@ -575,7 +575,7 @@ async fn add( let mut cfg = (*ctx.cfg()).clone(); let url = &feed::expand_input(url); // Includes feeds derived from an OPML, or the same show could be added twice. - if let Some(existing) = subscriptions(ctx)?.iter().find(|s| feed::same_feed(&s.cfg.url, url)) { + if let Some(existing) = subscriptions(ctx).await?.iter().find(|s| feed::same_feed(&s.cfg.url, url)) { anyhow::bail!("already subscribed as {:?}", existing.id); } let id = add_one(ctx, &mut cfg, url, folder, keywords).await?; @@ -627,13 +627,13 @@ pub async fn add_one( // Slugs must be unique across derived feeds too, or a new feed can collide with one // an OPML already introduced. - let mut taken: std::collections::BTreeMap<String, config::Feed> = subscriptions(ctx)? + let mut taken: std::collections::BTreeMap<String, config::Feed> = subscriptions(ctx).await? .into_iter() .map(|s| (s.id, s.cfg)) .collect(); // A removed feed keeps its rows, so its id is only free again for the same feed: re-adding // it gets its history back, and a different feed does not inherit someone else's. - for (id, other) in ctx.db.feed_urls()? { + for (id, other) in ctx.db.feed_urls().await? { if !feed::same_feed(&other, url) { taken.entry(id).or_insert_with(|| probe.clone()); } @@ -656,7 +656,7 @@ async fn rm(ctx: &Ctx, config_path: &std::path::Path, feed: &str) -> Result<()> if cfg.feeds.remove(feed).is_none() { // Derived from an OPML: drop it here, though the subscription will list it again // on the next read unless the OPML itself goes. - ctx.db.drop_managed(feed)?; + ctx.db.drop_managed(feed).await?; println!("removed {feed}; it came from an OPML subscription and may return on the next read"); return Ok(()); } @@ -703,7 +703,7 @@ pub async fn subscribe_opml( let mut found = vec![]; collect_outlines(&doc.body.outlines, &mut found); - let known = subscriptions(ctx)?; + let known = subscriptions(ctx).await?; let mut cfg = (*ctx.cfg()).clone(); let mut ids = vec![]; let mut grew = false; @@ -770,7 +770,7 @@ pub fn collect_outlines(outlines: &[opml::Outline], out: &mut Vec<(String, Strin } } -fn export(ctx: &Ctx, file: &std::path::Path) -> Result<()> { +async fn export(ctx: &Ctx, file: &std::path::Path) -> Result<()> { let mut doc = opml::OPML::default(); doc.head = Some(opml::Head { title: Some("ipx subscriptions".into()), @@ -779,7 +779,7 @@ fn export(ctx: &Ctx, file: &std::path::Path) -> Result<()> { for (id, feed) in &ctx.cfg().feeds { let title = ctx .db - .feed_summary(id) + .feed_summary(id).await .ok() .and_then(|s| s.title) .unwrap_or_else(|| id.clone()); @@ -791,14 +791,14 @@ fn export(ctx: &Ctx, file: &std::path::Path) -> Result<()> { Ok(()) } -fn list(ctx: &Ctx, config_path: &std::path::Path) -> Result<()> { +async fn list(ctx: &Ctx, config_path: &std::path::Path) -> Result<()> { let cfg = ctx.cfg(); if cfg.feeds.is_empty() { println!("No feeds configured in {}", config_path.display()); return Ok(()); } for (id, feed) in &cfg.feeds { - let s = ctx.db.feed_summary(id)?; + let s = ctx.db.feed_summary(id).await?; println!("{id} {}", s.title.as_deref().unwrap_or("-")); println!(" url {}", feed.url); println!(" last checked {}", ago(s.last_checked)); @@ -832,7 +832,7 @@ async fn reap(ctx: &Ctx, dry_run: bool, standalone: bool) -> Result<()> { async fn fetch(ctx: &Arc<Ctx>, only: Option<&str>, force: bool) -> Result<()> { let cfg = ctx.cfg(); - let subs = subscriptions(ctx)?; + let subs = subscriptions(ctx).await?; if let Some(id) = only && !subs.iter().any(|s| s.id == id) { @@ -843,7 +843,7 @@ async fn fetch(ctx: &Arc<Ctx>, only: Option<&str>, force: bool) -> Result<()> { let mut fresh: Vec<String> = vec![]; for sub in subs.iter().filter(|s| only.is_none_or(|o| o == s.id)) { let (id, feed_cfg) = (&sub.id, &sub.cfg); - let state = ctx.db.http_state(id)?; + let state = ctx.db.http_state(id).await?; if !force && let Some(last) = state.last_checked { let due = last + due_after(&cfg, feed_cfg, state.ttl_mins) as i64; @@ -890,20 +890,20 @@ async fn fetch(ctx: &Arc<Ctx>, only: Option<&str>, force: bool) -> Result<()> { // One bad feed must not end the scan. let msg = format!("{e:#}"); ctx.out.emit(Event::FeedError { feed: id.clone(), msg: msg.clone() }); - ctx.db.set_feed_error(id, &feed_cfg.url, &msg)?; + ctx.db.set_feed_error(id, &feed_cfg.url, &msg).await?; } } } // Feeds a subscribed OPML just introduced: scan them now, in this run. if !fresh.is_empty() { - let subs = subscriptions(ctx)?; + let subs = subscriptions(ctx).await?; for id in &fresh { let Some(feed_cfg) = subs.iter().find(|s| &s.id == id).map(|s| &s.cfg) else { continue; }; scanned += 1; ctx.out.emit(Event::FeedStart { feed: id.clone() }); - let state = ctx.db.http_state(id)?; + let state = ctx.db.http_state(id).await?; match scan_one(ctx, id, feed_cfg, &state).await { Ok(Outcome::Feed(s)) => ctx.out.emit(Event::FeedDone { feed: id.clone(), @@ -916,7 +916,7 @@ async fn fetch(ctx: &Arc<Ctx>, only: Option<&str>, force: bool) -> Result<()> { Err(e) => { let msg = format!("{e:#}"); ctx.out.emit(Event::FeedError { feed: id.clone(), msg: msg.clone() }); - ctx.db.set_feed_error(id, &feed_cfg.url, &msg)?; + ctx.db.set_feed_error(id, &feed_cfg.url, &msg).await?; } } } @@ -938,7 +938,7 @@ pub struct Sub { /// /// A derived feed borrows its parent's settings wholesale. That is why it needs no config /// entry -- there is nothing to store but its URL and where it came from. -pub fn subscriptions(ctx: &Ctx) -> Result<Vec<Sub>> { +pub async fn subscriptions(ctx: &Ctx) -> Result<Vec<Sub>> { let cfg = ctx.cfg(); let mut out: Vec<Sub> = cfg .feeds @@ -946,7 +946,7 @@ pub fn subscriptions(ctx: &Ctx) -> Result<Vec<Sub>> { .map(|(id, f)| Sub { id: id.clone(), cfg: f.clone(), managed: false }) .collect(); - for m in ctx.db.managed_feeds()? { + for m in ctx.db.managed_feeds().await? { if cfg.feeds.contains_key(&m.id) { continue; // promoted to config at some point; that entry wins } @@ -957,10 +957,10 @@ pub fn subscriptions(ctx: &Ctx) -> Result<Vec<Sub>> { // skip them here regardless so a row that slips through is never scanned. continue; } - let base = parent - .and_then(|p| p.folder.clone()) - .or_else(|| ctx.db.feed_summary(&m.group_id).ok().and_then(|s| s.title)) - .unwrap_or_else(|| m.group_id.clone()); + let base = match parent.and_then(|p| p.folder.clone()) { + Some(folder) => folder, + None => ctx.db.feed_summary(&m.group_id).await.ok().and_then(|s| s.title).unwrap_or_else(|| m.group_id.clone()), + }; let title = m.title.clone().unwrap_or_else(|| m.id.clone()); out.push(Sub { id: m.id.clone(), @@ -994,15 +994,15 @@ pub fn subscriptions(ctx: &Ctx) -> Result<Vec<Sub>> { /// derived any more, so it is only unmanaged. pub async fn retire_group(ctx: &Ctx, parent_id: &str) -> Result<()> { let cfg = ctx.cfg(); - for m in ctx.db.managed_feeds()?.into_iter().filter(|m| m.group_id == parent_id) { + for m in ctx.db.managed_feeds().await?.into_iter().filter(|m| m.group_id == parent_id) { if cfg.feeds.contains_key(&m.id) { // Scanned from its config entry and still read. Dropped as derived, its stored // entries would go with it: davewiner's 11 were promoted without being unmanaged. - ctx.db.unmanage(&m.id)?; + ctx.db.unmanage(&m.id).await?; } else if ctx.db.downloaded_count(&m.id).await.unwrap_or(1) > 0 { - ctx.db.set_orphaned(&m.id, true)?; + ctx.db.set_orphaned(&m.id, true).await?; } else { - ctx.db.drop_managed(&m.id)?; + ctx.db.drop_managed(&m.id).await?; } } Ok(()) @@ -1014,7 +1014,7 @@ pub async fn retire_group(ctx: &Ctx, parent_id: &str) -> Result<()> { /// most of the ones stored. async fn retire_stranded(ctx: &Ctx) -> Result<usize> { let cfg = ctx.cfg(); - let before = ctx.db.managed_feeds()?; + let before = ctx.db.managed_feeds().await?; let stranded: std::collections::BTreeSet<&str> = before .iter() .map(|m| m.group_id.as_str()) @@ -1023,7 +1023,7 @@ async fn retire_stranded(ctx: &Ctx) -> Result<usize> { for group in stranded { retire_group(ctx, group).await?; } - Ok(before.len() - ctx.db.managed_feeds()?.len()) + Ok(before.len() - ctx.db.managed_feeds().await?.len()) } /// Seconds to wait before re-checking a feed. @@ -1068,19 +1068,19 @@ async fn scan_one( if feed::is_patreon_creator(&feed_cfg.url) { match feed::patreon_shows(&ctx.client, &feed_cfg.url).await { Ok((name, shows)) if shows.len() > 1 => { - ctx.db.touch_feed(id, &feed_cfg.url)?; + ctx.db.touch_feed(id, &feed_cfg.url).await?; if let Some(name) = name { - ctx.db.set_title(id, &name)?; + ctx.db.set_title(id, &name).await?; } // Read as one feed before it was split, it listed every show's items in one // heap. The items go; its files and read state move to each show as the show // lists them (`Db::adopt`), so no show comes up empty for want of a URL. - ctx.db.clear_entries(id)?; + ctx.db.clear_entries(id).await?; return sync_group(ctx, id, feed_cfg, &shows).await; } Ok(_) => {} // One show: the creator's feed is that show. // Already split: keep the shows it has rather than read the creator as one heap. - Err(e) if ctx.db.managed_feeds()?.iter().any(|m| m.group_id == id) => return Err(e), + Err(e) if ctx.db.managed_feeds().await?.iter().any(|m| m.group_id == id) => return Err(e), Err(e) => tracing::warn!( feed = id, error = %format!("{e:#}"), @@ -1101,29 +1101,29 @@ async fn scan_one( // from backup, a manual edit, a cleanup that removed entries. Believe the database over // the validator: drop it and ask again, or the feed stays empty until the publisher // happens to change something. - if matches!(fetched, feed::Fetched::NotModified) && ctx.db.feed_summary(id)?.entries == 0 { + if matches!(fetched, feed::Fetched::NotModified) && ctx.db.feed_summary(id).await?.entries == 0 { tracing::info!(feed = id, "not modified, but nothing stored; refetching without the validator"); - ctx.db.clear_validators(id)?; + ctx.db.clear_validators(id).await?; fetched = feed::fetch(&ctx.client, feed_cfg, None, None).await?; } let (bytes, etag, last_modified) = match fetched { feed::Fetched::NotModified => { - ctx.db.touch_feed(id, &feed_cfg.url)?; + ctx.db.touch_feed(id, &feed_cfg.url).await?; return Ok(Outcome::NotModified); } feed::Fetched::Body { bytes, etag, last_modified } => (bytes, etag, last_modified), }; if bytes.iter().all(u8::is_ascii_whitespace) { - ctx.db.touch_feed(id, &feed_cfg.url)?; + ctx.db.touch_feed(id, &feed_cfg.url).await?; return Ok(Outcome::Empty); } // A subscribed OPML is a list of feeds, not a feed. The original matched on a ".opml" // URL; sniffing the body also catches one served from a URL without that extension. if feed::is_opml(&bytes) { - ctx.db.touch_feed(id, &feed_cfg.url)?; + ctx.db.touch_feed(id, &feed_cfg.url).await?; return sync_opml(ctx, id, feed_cfg, &bytes).await; } @@ -1137,7 +1137,7 @@ async fn scan_one( parsed.ttl_mins, parsed.image.as_deref(), parsed.category.as_deref(), - )?; + ).await?; let policy = policy_for(ctx, id, feed_cfg).await?; if let Some(parent) = &feed_cfg.group { @@ -1146,16 +1146,16 @@ async fn scan_one( .iter() .flat_map(|e| e.enclosures.iter().map(move |x| (e.guid.as_str(), x.url.as_str()))) .collect(); - ctx.db.adopt(parent, id, &listed)?; + ctx.db.adopt(parent, id, &listed).await?; } // Verdicts are recorded in `state`, so the download queue below is just "everything still // pending". A filter's verdict is looked at again on every scan, though: made once, at // discovery, it outlived the setting behind it, and allowing explicit items afterwards // changed nothing however often the feed was scanned. - let skipped = ctx.db.skipped_by_filter(id)?; + let skipped = ctx.db.skipped_by_filter(id).await?; let mut scan = Scan::default(); for entry in &parsed.entries { - if ctx.db.record_entry(id, entry)? { + if ctx.db.record_entry(id, entry).await? { scan.new_entries += 1; } for enc in &entry.enclosures { @@ -1263,7 +1263,7 @@ async fn sync_opml( ) -> Result<Outcome> { let listed = feed::parse_opml(bytes)?; if let Some(title) = feed::opml_title(bytes) { - ctx.db.set_title(parent_id, &title)?; + ctx.db.set_title(parent_id, &title).await?; } sync_group(ctx, parent_id, parent, &listed).await } @@ -1282,13 +1282,13 @@ async fn sync_group( listed: &[(String, String)], ) -> Result<Outcome> { let cfg = ctx.cfg(); - let existing = ctx.db.managed_feeds()?; + let existing = ctx.db.managed_feeds().await?; let mut added = vec![]; for (title, url) in listed { // Already known, whether derived or promoted into the config. if let Some(m) = existing.iter().find(|m| &m.url == url) { - ctx.db.upsert_managed(&m.id, url, title, parent_id)?; + ctx.db.upsert_managed(&m.id, url, title, parent_id).await?; continue; } // A Patreon show you added by hand may be spelled differently from the one listed. @@ -1296,7 +1296,7 @@ async fn sync_group( continue; } // A removed feed keeps its rows, so its id is only free again for the same feed. - let known = ctx.db.feed_urls()?; + let known = ctx.db.feed_urls().await?; let taken: std::collections::BTreeMap<String, config::Feed> = cfg .feeds .keys() @@ -1306,7 +1306,7 @@ async fn sync_group( .map(|id| (id.clone(), parent.clone())) .collect(); let id = config::unique_slug(title, &taken); - ctx.db.upsert_managed(&id, url, title, parent_id)?; + ctx.db.upsert_managed(&id, url, title, parent_id).await?; added.push(id); } @@ -1314,7 +1314,7 @@ async fn sync_group( // subscription means. Their own feeds are untouched. for id in ctx .db - .managed_feeds()? + .managed_feeds().await? .iter() .filter(|m| m.group_id == parent_id) .map(|m| m.id.clone()) @@ -1336,11 +1336,11 @@ async fn sync_group( } if ctx.db.downloaded_count(&m.id).await.unwrap_or(1) > 0 { // Never orphan a downloaded file: keep the feed and say why in the UI. - ctx.db.set_orphaned(&m.id, true)?; + ctx.db.set_orphaned(&m.id, true).await?; kept += 1; tracing::info!(feed = %m.id, "dropped from the OPML but has downloads; keeping it"); } else { - ctx.db.drop_managed(&m.id)?; + ctx.db.drop_managed(&m.id).await?; removed += 1; tracing::info!(feed = %m.id, "dropped from the OPML with nothing downloaded; removed"); } @@ -1546,7 +1546,7 @@ async fn download_one(ctx: &Arc<Ctx>, id: i64) -> Result<()> { } // Must look through the derived feeds too: anything inside an OPML subscription has // no config entry, so a config-only lookup called every one of them "unsubscribed". - let subs = subscriptions(ctx)?; + let subs = subscriptions(ctx).await?; let feed_cfg = subs .iter() .find(|s| s.id == enc.feed_id) @@ -1554,7 +1554,7 @@ async fn download_one(ctx: &Arc<Ctx>, id: i64) -> Result<()> { .ok_or_else(|| anyhow::anyhow!("enclosure {id} belongs to unsubscribed feed {:?}", enc.feed_id))?; let feed_cfg = &feed_cfg; - let title = ctx.db.feed_summary(&enc.feed_id)?.title; + let title = ctx.db.feed_summary(&enc.feed_id).await?.title; let folder = download::folder_for(&cfg, &enc.feed_id, feed_cfg, title.as_deref()); let dest_dir = cfg.general.download_dir.join(&folder); @@ -1743,9 +1743,9 @@ mod tests { // davewiner: the OPML subscription left config.toml, but its 922 derived rows // stayed in the database and kept being scanned under the no-parent fallback. let ctx = test_ctx(config::Config::default()).await; - ctx.db.upsert_managed("child", "http://x/child.xml", "Child", "gone-opml").unwrap(); + ctx.db.upsert_managed("child", "http://x/child.xml", "Child", "gone-opml").await.unwrap(); assert!( - subscriptions(&ctx).unwrap().iter().all(|s| s.id != "child"), + subscriptions(&ctx).await.unwrap().iter().all(|s| s.id != "child"), "a derived feed whose parent is gone from config must not be scanned" ); } @@ -1753,18 +1753,18 @@ mod tests { #[tokio::test] async fn retiring_a_group_drops_what_was_never_downloaded_and_orphans_the_rest() { let ctx = test_ctx(config::Config::default()).await; - ctx.db.upsert_managed("empty", "http://x/empty.xml", "Empty", "parent").unwrap(); - ctx.db.upsert_managed("has-file", "http://x/has-file.xml", "Has File", "parent").unwrap(); + ctx.db.upsert_managed("empty", "http://x/empty.xml", "Empty", "parent").await.unwrap(); + ctx.db.upsert_managed("has-file", "http://x/has-file.xml", "Has File", "parent").await.unwrap(); let enc = feed::Enclosure { url: "http://x/ep.mp3".into(), mime: None, length: None }; ctx.db.record_enclosure("has-file", "g1", &enc).await.unwrap(); ctx.db.mark_downloaded(&enc.url, std::path::Path::new("/downloads/ep.mp3"), 1).await.unwrap(); retire_group(&ctx, "parent").await.unwrap(); - let managed = ctx.db.managed_feeds().unwrap(); + let managed = ctx.db.managed_feeds().await.unwrap(); assert!(!managed.iter().any(|m| m.id == "empty"), "nothing downloaded, so it is forgotten"); assert!(managed.iter().any(|m| m.id == "has-file"), "has a file on disk, so it is kept"); - assert!(ctx.db.feed_summary("has-file").unwrap().orphaned, "and flagged as orphaned"); + assert!(ctx.db.feed_summary("has-file").await.unwrap().orphaned, "and flagged as orphaned"); } #[tokio::test] @@ -1775,15 +1775,15 @@ mod tests { cfg.feeds.insert("promoted".into(), feed()); cfg.feeds.insert("live-opml".into(), feed()); let ctx = test_ctx(cfg).await; - ctx.db.upsert_managed("promoted", "http://x/p.xml", "Promoted", "gone-opml").unwrap(); - ctx.db.upsert_managed("empty", "http://x/e.xml", "Empty", "gone-opml").unwrap(); - ctx.db.upsert_managed("listed", "http://x/l.xml", "Listed", "live-opml").unwrap(); - ctx.db.record_entry("promoted", &feed::Entry { guid: "g1".into(), ..Default::default() }).unwrap(); + ctx.db.upsert_managed("promoted", "http://x/p.xml", "Promoted", "gone-opml").await.unwrap(); + ctx.db.upsert_managed("empty", "http://x/e.xml", "Empty", "gone-opml").await.unwrap(); + ctx.db.upsert_managed("listed", "http://x/l.xml", "Listed", "live-opml").await.unwrap(); + ctx.db.record_entry("promoted", &feed::Entry { guid: "g1".into(), ..Default::default() }).await.unwrap(); assert_eq!(retire_stranded(&ctx).await.unwrap(), 2, "empty dropped, promoted unmanaged"); - let managed: Vec<String> = ctx.db.managed_feeds().unwrap().into_iter().map(|m| m.id).collect(); + let managed: Vec<String> = ctx.db.managed_feeds().await.unwrap().into_iter().map(|m| m.id).collect(); assert_eq!(managed, ["listed"], "a group still in config is left alone"); - assert_eq!(ctx.db.feed_summary("promoted").unwrap().entries, 1, "its entries survive"); + assert_eq!(ctx.db.feed_summary("promoted").await.unwrap().entries, 1, "its entries survive"); } } diff --git a/src/retention.rs b/src/retention.rs index a6e81e4..e58f2eb 100644 --- a/src/retention.rs +++ b/src/retention.rs @@ -66,7 +66,7 @@ pub async fn run(cfg: &Config, db: &Db, dry_run: bool) -> Result<Report> { report.bytes_freed += remove(db, c, dry_run).await?; } if !dry_run { - report.entries_pruned = db.prune_entries(cutoff)?; + report.entries_pruned = db.prune_entries(cutoff).await?; } } @@ -173,7 +173,7 @@ mod tests { (2, 'f', 'half', 'u2', '/tmp/half', 10, 'done', 20), (3, 'f', 'unread', 'u3', '/tmp/unread', 10, 'done', 30), (4, 'f', 'read', 'u4', '/tmp/read', 10, 'done', 40);", - ) + ).await .unwrap(); let got: Vec<i64> = db.reap_candidates().await.unwrap().iter().map(|c| c.id).collect(); @@ -198,9 +198,9 @@ mod tests { ('f', 'recent', 900); INSERT INTO enclosures (id, feed_id, guid, url, path, state) VALUES (1, 'f', 'has-file', 'u1', '/tmp/x', 'done');", - ) + ).await .unwrap(); - assert_eq!(db.prune_entries(500).unwrap(), 1, "only the old, fileless, unflagged one"); + assert_eq!(db.prune_entries(500).await.unwrap(), 1, "only the old, fileless, unflagged one"); } } diff --git a/src/web.rs b/src/web.rs index 1c897d5..f40ca22 100644 --- a/src/web.rs +++ b/src/web.rs @@ -672,7 +672,7 @@ async fn feeds( let cfg = state.ctx.cfg(); // Config entries plus the feeds derived from OPML subscriptions -- the catalogue. // What comes back is only the part of it this person subscribes to. - let subs = crate::subscriptions(&state.ctx)?; + let subs = crate::subscriptions(&state.ctx).await?; let mine: std::collections::HashMap<String, crate::db::Sub> = state .ctx .db @@ -689,8 +689,8 @@ async fn feeds( // the same fallback the scanner uses (`Db::subscribers`). let up = feed.group.as_deref().and_then(|g| mine.get(g)); let Some(mine) = mine.get(id) else { continue }; - let s = state.ctx.db.feed_summary(id)?; - let st = state.ctx.db.http_state(id)?; + let s = state.ctx.db.feed_summary(id).await?; + let st = state.ctx.db.http_state(id).await?; out.push(FeedRow { id: id.clone(), url: feed.url.clone(), @@ -806,7 +806,7 @@ async fn popular(state: &WebState, user_id: i64) -> Result<Vec<PopularRow>> { db.subscriptions_for(user_id).await?.into_iter().map(|s| s.feed_id).collect(); let counts = db.subscriber_counts().await?; let media = db.media_feeds().await?; - let catalogue = crate::subscriptions(&state.ctx)?; + let catalogue = crate::subscriptions(&state.ctx).await?; let by_id: std::collections::HashMap<&str, &crate::config::Feed> = catalogue.iter().map(|s| (s.id.as_str(), &s.cfg)).collect(); let is_folder: std::collections::HashSet<&str> = @@ -823,7 +823,7 @@ async fn popular(state: &WebState, user_id: i64) -> Result<Vec<PopularRow>> { { continue; } - let sum = db.feed_summary(&s.id)?; + let sum = db.feed_summary(&s.id).await?; let subscribed = mine.contains(&s.id); out.push(PopularRow { id: s.id.clone(), @@ -1219,7 +1219,7 @@ async fn add_feed( let url = crate::feed::expand_input(&body.url); // Someone else may already have it. Then adding costs nothing: no second fetch, no // second copy on disk, just another name against the same feed. - if let Some(existing) = crate::subscriptions(&state.ctx)? + if let Some(existing) = crate::subscriptions(&state.ctx).await? .into_iter() .find(|s| crate::feed::same_feed(&s.cfg.url, &url)) { @@ -1341,13 +1341,13 @@ async fn patch_feed( // Derived feeds have no config entry. Editing one is the moment it earns a real // entry: promote it, so the config holds your decisions and nothing else. if !cfg.feeds.contains_key(&id) { - let subs = crate::subscriptions(&state.ctx)?; + let subs = crate::subscriptions(&state.ctx).await?; let found = subs .iter() .find(|s| s.id == id) .ok_or_else(|| ApiError::bad_request(format!("no feed with id {id:?}")))?; cfg.feeds.insert(id.clone(), found.cfg.clone()); - state.ctx.db.unmanage(&id)?; + state.ctx.db.unmanage(&id).await?; } let checked = match &body.url { @@ -1388,7 +1388,7 @@ async fn patch_feed( if url_changed { // Refreshing a rotated auth token is the common case; entries and download history // are keyed by feed id, so they survive the change. - state.ctx.db.clear_validators(&id)?; + state.ctx.db.clear_validators(&id).await?; } Ok(StatusCode::NO_CONTENT) } @@ -1401,7 +1401,7 @@ async fn remove_feed( // Unsubscribing is personal: it takes the feed off your list and leaves everyone // else's alone. state.ctx.db.unsubscribe(user.id, &id).await?; - for child in crate::subscriptions(&state.ctx)? + for child in crate::subscriptions(&state.ctx).await? .iter() .filter(|s| s.cfg.group.as_deref() == Some(id.as_str())) { @@ -1417,7 +1417,7 @@ async fn remove_feed( if cfg.feeds.remove(&id).is_none() { // A derived feed: forget it here, though the OPML will list it again on the next // read unless you unsubscribe from the OPML itself. - state.ctx.db.drop_managed(&id)?; + state.ctx.db.drop_managed(&id).await?; return Ok(StatusCode::NO_CONTENT); } cfg.save(&state.config_path)?; @@ -1610,7 +1610,7 @@ async fn read_all( // A subscription's own row has no entries, so marking it read means everything under it. let mut ids = vec![id.clone()]; ids.extend( - crate::subscriptions(&state.ctx)? + crate::subscriptions(&state.ctx).await? .into_iter() .filter(|s| s.cfg.group.as_deref() == Some(id.as_str())) .map(|s| s.id), @@ -1676,7 +1676,7 @@ async fn export_opml( }), ..Default::default() }; - for s in crate::subscriptions(&state.ctx)? { + for s in crate::subscriptions(&state.ctx).await? { // A feed from an OPML subscription comes back with the OPML itself. if s.managed || !mine.contains(&s.id) { continue; @@ -1684,7 +1684,7 @@ async fn export_opml( let title = state .ctx .db - .feed_summary(&s.id) + .feed_summary(&s.id).await .ok() .and_then(|sum| sum.title) .unwrap_or_else(|| s.id.clone());