//! 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::{catalogue, enclosures, entries, feeds, sessions, settings, 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; /// A pool of connections, where there was one connection behind a global lock. pub struct Db { orm: sea_orm::DatabaseConnection, /// A test's database file, removed when the test is done with it. #[cfg(test)] tmp: Option, } #[cfg(test)] impl Drop for Db { fn drop(&mut self) { if let Some(p) = &self.tmp { for ext in ["", "-wal", "-shm"] { let _ = std::fs::remove_file(format!("{}{ext}", p.display())); } } } } /// 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. /// /// 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, 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), schema.create_table_from_entity(catalogue::Entity), schema.create_table_from_entity(settings::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))", ] { orm.execute_unprepared(sql).await.with_context(|| sql.to_owned())?; } Ok(()) } /// One table's rows, a page at a time in primary-key order, from one database to another. async fn copy_table(from: &sea_orm::DatabaseConnection, to: &impl ConnectionTrait) -> Result where E: EntityTrait, E::Model: sea_orm::IntoActiveModel + Send + Sync, E::ActiveModel: ActiveModelTrait + Send, { use sea_orm::{IntoActiveModel, Iterable, PrimaryKeyToColumn}; let mut query = E::find(); for key in E::PrimaryKey::iter() { query = query.order_by_asc(key.into_column()); } let mut pages = query.paginate(from, 1000); let mut n = 0; while let Some(rows) = pages.fetch_and_next().await? { n += rows.len() as u64; // reset_all: every column written, the primary key included, not just the changed ones. E::insert_many(rows.into_iter().map(|m| m.into_active_model().reset_all())) .exec_without_returning(to) .await?; } Ok(n) } /// Where the database is: IPX_DATABASE_URL, a postgres:// URL, when it is set; otherwise the /// SQLite file in the data directory, as it has always been. pub fn location() -> String { std::env::var("IPX_DATABASE_URL") .ok() .filter(|u| !u.trim().is_empty()) .unwrap_or_else(|| crate::config::data_dir().join("state.db").display().to_string()) } /// The Postgres connection option that keeps notices from being sent at all; see `connect`. const QUIET: &str = "options=-c%20client_min_messages%3Dwarning"; /// A URL fit for a log or an error: the password taken out. fn redact(url: &str) -> String { match (url.find("://"), url.rfind('@')) { (Some(s), Some(at)) if at > s => match url[s + 3..at].find(':') { Some(c) => format!("{}:***{}", &url[..s + 3 + c], &url[at..]), None => url.to_owned(), }, _ => url.to_owned(), } } fn is_postgres(location: &str) -> bool { location.starts_with("postgres://") || location.starts_with("postgresql://") } /// 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(location: &str) -> Result { let mut opts = sea_orm::ConnectOptions::new(url_for(location)); opts.sqlx_logging(false); sea_orm::Database::connect(opts) .await .with_context(|| format!("opening {}", redact(location))) } /// The URL `connect` hands sqlx for a location: a SQLite file, created if missing, or a Postgres /// URL with notices turned off. fn url_for(location: &str) -> String { if !is_postgres(location) { format!("sqlite://{location}?mode=rwc") } else if location.contains("options=") { location.to_owned() // someone chose their own; theirs stands } else { // Warnings and up only. Postgres answers every CREATE ... IF NOT EXISTS on an existing // table with a notice, eleven on each open; sqlx logs each, no filter here wants them, // and tracing-subscriber's per-layer filters then swallowed the next line ipx logged // (0.3.23: "moved the feeds ... into the database" went missing that way). let sep = if location.contains('?') { '&' } else { '?' }; format!("{location}{sep}{QUIET}") } } /// One person's wants for one feed. `None` in a field means the feed's own setting stands. #[derive(Debug, Clone, Default)] pub struct Sub { pub feed_id: String, pub keywords: Option>, pub auto_download: Option, pub allow_explicit: Option, pub max_new_per_check: Option, } /// Someone who can sign in. `pass_hash` is None for an account that only ever arrives /// through the proxy. #[derive(Debug, Clone)] pub struct User { pub id: i64, pub name: String, pub pass_hash: Option, pub is_admin: bool, /// When the account was made and when it last signed in, for whoever maintains the server. pub created: Option, pub last_login: Option, } impl From for EncRow { fn from(e: enclosures::Model) -> Self { EncRow { id: e.id, feed_id: e.feed_id, guid: e.guid, url: e.url, mime: e.mime, length: e.length, path: e.path, state: e.state, last_error: e.last_error, } } } /// A subscription's keywords, stored as a JSON array. fn keywords(json: Option) -> Option> { json.and_then(|j| serde_json::from_str(&j).ok()) } impl From for Sub { fn from(s: subscriptions::Model) -> Self { Sub { feed_id: s.feed_id, keywords: keywords(s.keywords), auto_download: s.auto_download, allow_explicit: s.allow_explicit, max_new_per_check: s.max_new_per_check, } } } impl From for User { fn from(u: users::Model) -> Self { User { id: u.id, name: u.name, pass_hash: u.pass_hash, is_admin: u.is_admin, created: u.created, last_login: u.last_login, } } } /// A feed derived from an OPML subscription rather than written into the config. #[derive(Debug, Clone)] pub struct Managed { pub id: String, pub url: String, pub title: Option, pub group_id: String, } /// What `ipx list` shows next to each configured feed. #[derive(Debug, Default)] pub struct FeedSummary { pub title: Option, pub image: Option, pub category: Option, /// Came from a subscribed OPML that no longer lists it, but it has downloads, so it /// was kept rather than removed. pub orphaned: bool, pub last_checked: Option, pub last_error: Option, /// When this run of failures began; see the `error_since` column. pub error_since: Option, pub entries: i64, pub downloaded: i64, } impl Db { /// Opens the database at `location` (see `location()`): a postgres:// URL, or a SQLite file. pub async fn open(location: &str) -> Result { if !is_postgres(location) && let Some(dir) = Path::new(location).parent() { std::fs::create_dir_all(dir).with_context(|| format!("creating {}", dir.display()))?; } let orm = connect(location).await?; if !is_postgres(location) { // 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 { orm, #[cfg(test)] tmp: None, }) } /// A fresh, empty database for a test, its schema made from the entities as a real one's is. /// A SQLite file of its own (two connections to one ":memory:" would be two databases); or, /// with IPX_TEST_DATABASE_URL set, a Postgres schema of its own on that database, so the /// same tests prove the SQL on both. #[cfg(test)] pub async fn memory() -> Result { static N: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0); let n = N.fetch_add(1, std::sync::atomic::Ordering::Relaxed); let pid = std::process::id(); if let Some(url) = std::env::var("IPX_TEST_DATABASE_URL").ok().filter(|u| !u.is_empty()) { let admin = connect(&url).await?; // Schemas an earlier run left behind: anything of ours not made by this process. let old = admin .query_all_raw(Statement::from_string( admin.get_database_backend(), format!( r"SELECT nspname FROM pg_namespace WHERE nspname LIKE 'ipxt\_%' AND nspname NOT LIKE 'ipxt\_{pid}\_%'" ), )) .await?; for r in old { let name: String = r.try_get_by_index(0)?; admin.execute_unprepared(&format!("DROP SCHEMA IF EXISTS {name} CASCADE")).await?; } let schema = format!("ipxt_{pid}_{n}"); admin.execute_unprepared(&format!("CREATE SCHEMA {schema}")).await?; // Every connection in the pool starts in it, so each test sees only its own tables. let sep = if url.contains('?') { '&' } else { '?' }; let orm = connect(&format!("{url}{sep}options=-c%20search_path%3D{schema}%20-c%20client_min_messages%3Dwarning")) .await?; create_missing(&orm).await?; return Ok(Self { orm, tmp: None }); } let path = std::env::temp_dir().join(format!("ipx-test-{pid}-{n}.db")); let orm = connect(&path.display().to_string()).await?; create_missing(&orm).await?; Ok(Self { orm, tmp: Some(path) }) } #[cfg(test)] /// 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 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 = 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) } } /// Everything `fetch` needs to decide whether to poll a feed, and how. #[derive(Debug, Default)] pub struct HttpState { pub etag: Option, pub last_modified: Option, pub last_checked: Option, pub ttl_mins: Option, } impl Db { 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. #[allow(clippy::too_many_arguments)] pub async fn record_feed( &self, feed_id: &str, url: &str, title: Option<&str>, etag: Option<&str>, last_modified: Option<&str>, ttl_mins: Option, image: Option<&str>, category: Option<&str>, ) -> Result<()> { // category is taken as it comes, unlike title and image: a show that leaves a category // should leave the Directory's chip too. 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 url = excluded.url, title = coalesce(excluded.title, feeds.title), etag = excluded.etag, last_modified = excluded.last_modified, last_checked = excluded.last_checked, ttl_mins = excluded.ttl_mins, image = coalesce(excluded.image, feeds.image), category = excluded.category, last_error = NULL, error_since = NULL", 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 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", 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 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 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 last_checked = excluded.last_checked, last_error = excluded.last_error, error_since = coalesce(feeds.error_since, excluded.error_since)", vec![feed_id.into(), url.into(), now().into(), msg.into()], ) .await?; Ok(()) } /// Returns true when this entry had not been seen before. /// 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) } /// Returns true when this enclosure URL is new. False means we have downloaded it /// before, or deliberately reaped it -- either way it is not fetched again. pub async fn mark_downloaded(&self, url: &str, path: &std::path::Path, bytes: u64) -> Result<()> { self.exec( "UPDATE enclosures SET state = 'done', path = $2, bytes_done = $3, downloaded_at = $4, last_error = NULL WHERE url = $1", vec![url.into(), path.to_string_lossy().into_owned().into(), (bytes as i64).into(), now().into()], ) .await?; Ok(()) } /// The row stays -- a failed URL is still a URL we have seen. `state` says why it has /// no file, and a retry is an explicit act rather than something a rescan does silently. pub async fn mark_enclosure(&self, url: &str, state: &str, error: Option<&str>) -> Result<()> { self.exec( "UPDATE enclosures SET state = $2, last_error = $3 WHERE url = $1", vec![url.into(), state.into(), error.map(str::to_owned).into()], ) .await?; Ok(()) } pub async fn record_enclosure( &self, feed_id: &str, guid: &str, enc: &crate::feed::Enclosure, ) -> Result { let inserted = self .exec( "INSERT INTO enclosures (feed_id, guid, url, mime, length, state) VALUES ($1, $2, $3, $4, $5, 'pending') ON CONFLICT DO NOTHING", vec![ feed_id.into(), guid.into(), enc.url.clone().into(), enc.mime.clone().into(), enc.length.into(), ], ) .await?; Ok(inserted == 1) } } /// An enclosure waiting to be downloaded. #[derive(Debug)] pub struct Pending { pub id: i64, pub url: String, pub mime: Option, } impl Db { /// The download queue is the table, not the parse result: an enclosure held back by /// `max_new_per_check` is simply picked up by the next scan, in feed order. pub async fn pending(&self, feed_id: &str, limit: usize) -> Result> { // Newest first: a cap of 3 should mean the three latest episodes, not the three that // happen to have been recorded first. self.rows( "SELECT x.id, x.url, x.mime FROM enclosures x JOIN entries e ON e.feed_id = x.feed_id AND e.guid = x.guid WHERE x.feed_id = $1 AND x.state = 'pending' ORDER BY coalesce(e.published, e.first_seen) DESC, x.id DESC LIMIT $2", vec![feed_id.into(), (limit as i64).into()], ) .await? .iter() .map(|r| Ok(Pending { id: r.try_get("", "id")?, url: r.try_get("", "url")?, mime: r.try_get("", "mime")? })) .collect() } } /// A downloaded file the reaper may consider. #[derive(Debug, Clone, PartialEq)] pub struct Candidate { pub id: i64, pub url: String, pub path: String, pub bytes: i64, /// When it landed; the reaper works oldest-first. pub age_key: i64, pub read: bool, } impl Db { /// Files on disk that may be deleted to get back under quota: starred by nobody, /// with the ones everybody has finished going first, oldest first within each group. /// /// One file serves every subscriber, so both tests are about all of them: **anyone** /// starring it keeps it, and it only counts as read when **everyone** subscribed has /// read it. A file whose feed nobody subscribes to has no one left to keep it, so it /// sorts with the read ones. /// /// (The Python intended `read = 1 AND flagged = 0` but never achieved it -- a missing /// plistlib import and an `EntreiesData` typo meant the filter always threw.) pub async fn reap_candidates(&self) -> Result> { // Yes/no as true and false, not 1 and 0: Postgres types a bare 1 as a 32-bit integer // and will not hand it over as an i64. self.rows( "SELECT e.id, e.url, e.path, e.bytes_done, coalesce(e.downloaded_at, 0) AS age_key, CASE WHEN coalesce(readers.n, 0) >= coalesce(subs.n, 0) THEN true ELSE false END AS read FROM enclosures e LEFT JOIN (SELECT feed_id, count(*) AS n FROM subscriptions GROUP BY feed_id) subs ON subs.feed_id = e.feed_id LEFT JOIN (SELECT feed_id, guid, count(*) AS n FROM entry_state WHERE read GROUP BY feed_id, guid) readers ON readers.feed_id = e.feed_id AND readers.guid = e.guid WHERE e.path IS NOT NULL AND NOT EXISTS (SELECT 1 FROM entry_state s WHERE s.feed_id = e.feed_id AND s.guid = e.guid AND s.flagged) ORDER BY 6 DESC, coalesce(e.downloaded_at, 0) ASC, e.id ASC", vec![], ) .await? .iter() .map(|r| { Ok(Candidate { id: r.try_get("", "id")?, url: r.try_get("", "url")?, path: r.try_get("", "path")?, bytes: r.try_get("", "bytes_done")?, age_key: r.try_get("", "age_key")?, read: r.try_get("", "read")?, }) }) .collect() } /// The row survives the file: that is what stops a reaped episode being re-downloaded. pub async fn mark_reaped(&self, id: i64) -> Result<()> { self.exec("UPDATE enclosures SET state = 'reaped', path = NULL WHERE id = $1", vec![id.into()]).await?; Ok(()) } /// Rows claiming a file that is no longer there (someone deleted it by hand). pub async fn missing_files(&self) -> Result> { Ok(enclosures::Entity::find() .filter(enclosures::Column::Path.is_not_null()) .all(&self.orm) .await? .into_iter() .filter_map(|e| Some((e.id, e.path?))) .filter(|(_, p)| !std::path::Path::new(p).exists()) .collect()) } /// Old entries that never had a file, or no longer have one. Enclosure rows stay -- /// they are the dedupe history. 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. 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)", vec![], ) .await?; Ok(n as usize) } } impl Db { pub async fn unread_count(&self, user_id: i64, feed_id: &str) -> Result { let r = self .rows( "SELECT count(*) AS n FROM entries e LEFT JOIN entry_state s ON s.user_id = $2 AND s.feed_id = e.feed_id AND s.guid = e.guid WHERE e.feed_id = $1 AND NOT coalesce(s.read, false)", vec![feed_id.into(), user_id.into()], ) .await?; Ok(r.first().context("count(*) always returns a row")?.try_get("", "n")?) } /// (pending, downloaded) across all feeds, for the status command. pub async fn counts(&self) -> Result<(i64, i64)> { let pending = enclosures::Entity::find() .filter(enclosures::Column::State.eq("pending")) .count(&self.orm) .await?; let downloaded = enclosures::Entity::find() .filter(enclosures::Column::Path.is_not_null()) .count(&self.orm) .await?; Ok((pending as i64, downloaded as i64)) } } /// An entry plus its enclosures, for the web UI. #[derive(Debug, serde::Serialize)] pub struct EntryRow { pub guid: String, pub feed_id: String, pub title: Option, pub link: Option, pub published: Option, pub description: Option, pub read: bool, pub flagged: bool, pub image: Option, pub duration: Option, pub episode: Option, pub season: Option, pub position: i64, pub enclosures: Vec, } /// A statement's parameters, gathered as its SQL is written: each `p` binds a value and gives /// back its `$n`. Only what the SQL uses is bound -- Postgres refuses a parameter it cannot /// place, where rusqlite needed every one mentioned, which is what `?1 IS NULL` was for. #[derive(Default)] struct Args(Vec); impl Args { fn p(&mut self, v: impl Into) -> String { self.0.push(v.into()); format!("${}", self.0.len()) } } /// The FROM and WHERE that `entries_in` and `count_in` share: whose read state, which feeds /// (one, or every feed the person subscribes to), the filter, and a case-insensitive search of /// title and description. fn entries_from(a: &mut Args, user_id: i64, feed_id: Option<&str>, filter: Filter, search: Option<&str>) -> String { let user = a.p(user_id); let scope = match feed_id { Some(f) => format!("e.feed_id = {}", a.p(f)), None => format!("e.feed_id IN (SELECT feed_id FROM subscriptions WHERE user_id = {user})"), }; let search = match search.map(|q| q.trim().to_lowercase()).filter(|q| !q.is_empty()) { Some(q) => { let like = a.p(format!("%{q}%")); format!( "(lower(coalesce(e.title, '')) LIKE {like} OR lower(coalesce(e.description, '')) LIKE {like})" ) } None => "true".into(), }; format!( "FROM entries e LEFT JOIN entry_state s ON s.user_id = {user} AND s.feed_id = e.feed_id AND s.guid = e.guid WHERE {scope} AND {} AND {search}", filter.sql() ) } /// The item table's ORDER BY. The column name picks one of these fixed expressions, so nothing /// the caller sends reaches the query, and anything unrecognised is newest first. Ties fall back /// to newest first too, so a page boundary is stable across "Load more". /// /// ponytail: file type and size look at the item's first and largest file. The row shows the file /// it summarises, which is almost always that one; sort by that one if they ever disagree. /// /// Pinned items are not lifted above the rest: sorting by the pin column, or the Pinned tab, does /// that when it is wanted, and lifting them always was undone (issues #35 and #36). pub fn order_sql(col: &str, dir: &str) -> String { let expr = match col { "kept" => "coalesce(s.flagged, false)", "title" => "lower(coalesce(e.title, ''))", "feed" => "(SELECT lower(coalesce(f.title, f.id)) FROM feeds f WHERE f.id = e.feed_id)", "type" => "(SELECT min(x.mime) FROM enclosures x WHERE x.feed_id = e.feed_id AND x.guid = e.guid)", "size" => "(SELECT max(x.length) FROM enclosures x WHERE x.feed_id = e.feed_id AND x.guid = e.guid)", _ => "coalesce(e.published, e.first_seen)", }; // Where an item with no value goes, said outright: SQLite counts a NULL as the smallest value // and Postgres as the largest, so "largest first" on Postgres opened with every item that has // no file. NULLS FIRST going up and LAST going down keeps what SQLite did. let dir = if dir == "asc" { "ASC NULLS FIRST" } else { "DESC NULLS LAST" }; // The guid breaks what ties remain: SQLite's rowid did, and Postgres has none. format!("{expr} {dir}, coalesce(e.published, e.first_seen) DESC, e.guid DESC") } /// Which slice of a feed the UI is asking for. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum Filter { All, Unread, Downloaded, Flagged, /// Started (a saved playback position past the first few seconds) and short of the 90% /// where `markPlayed` in the UI calls it finished. Not `read`: opening an item marks it /// read, so filtering on that hid nearly every episode anyone had started. The length is the /// one this person's player measured where there is one (`set_position`), else the feed's; /// with neither, the episode counts as unfinished. /// Currently Listening, below Popular, is this filter on every feed at once. InProgress, } impl Filter { pub fn parse(s: &str) -> Self { match s { "unread" => Self::Unread, "downloaded" => Self::Downloaded, "flagged" => Self::Flagged, "in_progress" => Self::InProgress, _ => Self::All, } } /// The WHERE fragment for this filter. `e` is entries, and the EXISTS subquery is /// correlated against it. fn sql(self) -> &'static str { match self { // true and false, not 1 and 0: the columns are booleans on Postgres, and SQLite // reads true and false as 1 and 0. Self::All => "true", Self::Unread => "NOT coalesce(s.read, false)", Self::Flagged => "coalesce(s.flagged, false)", Self::Downloaded => { "EXISTS (SELECT 1 FROM enclosures x WHERE x.feed_id = e.feed_id AND x.guid = e.guid AND x.path IS NOT NULL)" } Self::InProgress => { "coalesce(s.position, 0) > 5 AND (coalesce(s.duration, e.duration, 0) = 0 OR s.position * 10 < coalesce(s.duration, e.duration) * 9)" } } } } #[derive(Debug, Clone, serde::Serialize)] pub struct EncRow { pub id: i64, pub feed_id: String, pub guid: String, pub url: String, pub mime: Option, pub length: Option, pub path: Option, pub state: String, pub last_error: Option, } impl Db { /// One page of entries, each with its enclosures attached: one feed's, or every feed the /// person subscribes to when `feed_id` is None (All Subscriptions). `search` matches title /// and description, case-insensitively. #[allow(clippy::too_many_arguments)] pub async fn entries_in( &self, user_id: i64, feed_id: Option<&str>, filter: Filter, search: Option<&str>, offset: i64, limit: i64, order: &str, ) -> Result> { let mut a = Args::default(); let from = entries_from(&mut a, user_id, feed_id, filter, search); let sql = format!( "SELECT e.guid, e.feed_id, e.title, e.link, e.published, e.description, coalesce(s.read, false) AS read, coalesce(s.flagged, false) AS flagged, e.image, coalesce(s.duration, e.duration) AS duration, e.episode, e.season, coalesce(s.position, 0) AS position {from} ORDER BY {order} LIMIT {} OFFSET {}", a.p(limit), a.p(offset) ); let mut rows = self .rows(&sql, a.0) .await? .iter() .map(|r| { Ok(EntryRow { guid: r.try_get("", "guid")?, feed_id: r.try_get("", "feed_id")?, title: r.try_get("", "title")?, link: r.try_get("", "link")?, published: r.try_get("", "published")?, description: r.try_get("", "description")?, read: r.try_get("", "read")?, flagged: r.try_get("", "flagged")?, image: r.try_get("", "image")?, duration: r.try_get("", "duration")?, episode: r.try_get("", "episode")?, season: r.try_get("", "season")?, position: r.try_get("", "position")?, enclosures: vec![], }) }) .collect::>>()?; if rows.is_empty() { return Ok(rows); } // Only the guids on this page, so a feed with thousands of entries stays cheap. A page // can span feeds, so each file is matched to its row by feed as well as guid, below. let encs = enclosures::Entity::find() .filter(enclosures::Column::Guid.is_in(rows.iter().map(|r| r.guid.clone()))) .order_by_asc(enclosures::Column::Id) .all(&self.orm) .await?; for enc in encs { if let Some(row) = rows.iter_mut().find(|r| r.guid == enc.guid && r.feed_id == enc.feed_id) { row.enclosures.push(EncRow::from(enc)); } } Ok(rows) } /// How many entries `entries_in` would page through, so the UI knows whether there is more. pub async fn count_in( &self, user_id: i64, feed_id: Option<&str>, filter: Filter, search: Option<&str>, ) -> Result { let mut a = Args::default(); let sql = format!("SELECT count(*) AS n {}", entries_from(&mut a, user_id, feed_id, filter, search)); let r = self.rows(&sql, a.0).await?; Ok(r.first().context("count(*) always returns a row")?.try_get("", "n")?) } /// Where playback got to, so it resumes there next time -- for this listener only. /// `duration` is the length their player measured, kept beside the position and preferred to /// the feed's for time left and for when an episode counts as finished. A feed can be minutes /// out: ReThinking's gave 41:23 for a 43:48 file, which said 0:08 left with 2:33 to play. /// Not written to `entries`, where every scan puts the feed's figure back, and kept per /// listener so one person's player never changes what anyone else sees. pub async fn set_position( &self, user_id: i64, feed_id: &str, guid: &str, secs: i64, duration: Option, ) -> Result<()> { // entry_state.duration, named in full: Postgres will not guess between it and excluded's. self.exec( "INSERT INTO entry_state (user_id, feed_id, guid, position, duration) VALUES ($1, $2, $3, $4, $5) ON CONFLICT (user_id, feed_id, guid) DO UPDATE SET position = excluded.position, duration = coalesce(excluded.duration, entry_state.duration)", vec![ user_id.into(), feed_id.into(), guid.into(), secs.max(0).into(), duration.filter(|d| *d > 0).into(), ], ) .await?; Ok(()) } // ---- the configuration: the catalogue and the server settings ---- /// The feeds and server settings, once the database holds them; None before, when they are /// still config.toml's. pub async fn stored_config( &self, ) -> Result)>> { let Some(general) = settings::Entity::find_by_id("general".to_owned()).one(&self.orm).await? else { return Ok(None); }; let stored = serde_json::from_str(&general.value).context("reading the stored server settings")?; let mut feeds = std::collections::BTreeMap::new(); for row in catalogue::Entity::find().all(&self.orm).await? { let feed = serde_json::from_str(&row.spec).with_context(|| format!("reading feed {}", row.id))?; feeds.insert(row.id, feed); } Ok(Some((stored, feeds))) } /// Takes the feeds and server settings from `cfg` (config.toml, as read) into a database that /// has none. False when it already had them: another process got there first, and its copy /// stands. One transaction, and the settings row goes in first, so two processes starting at /// once cannot both import. pub async fn import_config(&self, cfg: &crate::config::Config) -> Result { use sea_orm::TransactionTrait; let backend = self.orm.get_database_backend(); let tx = self.orm.begin().await?; let claimed = tx .execute_raw(Statement::from_sql_and_values( backend, "INSERT INTO settings (name, value) VALUES ('general', $1) ON CONFLICT DO NOTHING", vec![serde_json::to_string(&crate::config::Stored::of(cfg))?.into()], )) .await? .rows_affected(); if claimed == 0 { return Ok(false); // dropped, so rolled back } for (id, feed) in &cfg.feeds { catalogue::ActiveModel { id: Set(id.clone()), spec: Set(serde_json::to_string(feed)?) } .insert(&tx) .await?; } tx.commit().await?; Ok(true) } /// Writes the feeds and server settings as they now stand: what config.toml's save did. /// The whole catalogue at once, in one transaction, so a feed removed goes too. pub async fn store_config(&self, cfg: &crate::config::Config) -> Result<()> { use sea_orm::TransactionTrait; let backend = self.orm.get_database_backend(); let tx = self.orm.begin().await?; tx.execute_raw(Statement::from_sql_and_values( backend, "INSERT INTO settings (name, value) VALUES ('general', $1) ON CONFLICT (name) DO UPDATE SET value = excluded.value", vec![serde_json::to_string(&crate::config::Stored::of(cfg))?.into()], )) .await?; catalogue::Entity::delete_many().exec(&tx).await?; for (id, feed) in &cfg.feeds { catalogue::ActiveModel { id: Set(id.clone()), spec: Set(serde_json::to_string(feed)?) } .insert(&tx) .await?; } tx.commit().await?; Ok(()) } // ---- moving to another database ---- /// Copies every row of `from` into this database, which must be empty: the move from the /// SQLite file to Postgres (issue #18). One transaction, so a copy that fails part-way leaves /// nothing behind and can simply be run again. Returns each table's row count. pub async fn copy_from(&self, from: &Db) -> Result> { use crate::entity::*; use sea_orm::TransactionTrait; let held = users::Entity::find().count(&self.orm).await? + feeds::Entity::find().count(&self.orm).await? + entries::Entity::find().count(&self.orm).await?; anyhow::ensure!(held == 0, "the database being copied into already holds rows; it has to be empty"); let tx = self.orm.begin().await?; // Users first: the tables that belong to a person refer to them. let counts = vec![ ("users", copy_table::(&from.orm, &tx).await?), ("feeds", copy_table::(&from.orm, &tx).await?), ("entries", copy_table::(&from.orm, &tx).await?), ("enclosures", copy_table::(&from.orm, &tx).await?), ("subscriptions", copy_table::(&from.orm, &tx).await?), ("entry_state", copy_table::(&from.orm, &tx).await?), ("sessions", copy_table::(&from.orm, &tx).await?), ("catalogue", copy_table::(&from.orm, &tx).await?), ("settings", copy_table::(&from.orm, &tx).await?), ]; if self.orm.get_database_backend() == sea_orm::DbBackend::Postgres { // The copied ids came with the rows; the counters that hand out new ones start past // them, or the next account or file would collide with one copied. for table in ["users", "enclosures"] { tx.execute_unprepared(&format!( "SELECT setval(pg_get_serial_sequence('{table}', 'id'), \ coalesce((SELECT max(id) FROM {table}), 0) + 1, false)" )) .await?; } } tx.commit().await?; Ok(counts) } // ---- hand-written SQL ---- // // For what reads better as SQL than as a query builder: joins, sums, upserts. Written to // run on SQLite and Postgres alike -- $1-style parameters, ON CONFLICT, CASE WHEN on a // yes/no column rather than comparing it to 1 -- since both run it (issue #18). fn stmt(&self, sql: &str, values: Vec) -> Statement { Statement::from_sql_and_values(self.orm.get_database_backend(), sql, values) } /// Runs a statement; how many rows it changed. async fn exec(&self, sql: &str, values: Vec) -> Result { Ok(self.orm.execute_raw(self.stmt(sql, values)).await?.rows_affected()) } async fn rows(&self, sql: &str, values: Vec) -> Result> { Ok(self.orm.query_all_raw(self.stmt(sql, values)).await?) } /// The first admin starts subscribed to the whole catalogue: whoever wrote config.toml meant /// to read those feeds, and without this a fresh install signs in to an empty sidebar. Runs /// only while nobody subscribes to anything, so an unsubscribe is never undone. pub async fn adopt_catalogue(&self, user_id: i64, catalogue: &[String]) -> Result { if subscriptions::Entity::find().count(&self.orm).await? > 0 { return Ok(0); } for id in catalogue { self.subscribe(user_id, id).await?; } // Feeds that exist only in the database (OPML children) count too. SQLite wants the // WHERE to tell the SELECT from the ON CONFLICT; Postgres does not mind it. self.exec( "INSERT INTO subscriptions (user_id, feed_id) SELECT $1, id FROM feeds WHERE true ON CONFLICT DO NOTHING", vec![user_id.into()], ) .await?; Ok(catalogue.len()) } // ---- subscriptions ---- /// What this person wants from a feed. Absent means they do not subscribe at all. pub async fn subscription(&self, user_id: i64, feed_id: &str) -> Result> { Ok(subscriptions::Entity::find_by_id((user_id, feed_id.to_owned())) .one(&self.orm) .await? .map(Sub::from)) } /// Every feed this person subscribes to, with their settings. pub async fn subscriptions_for(&self, user_id: i64) -> Result> { Ok(subscriptions::Entity::find() .filter(subscriptions::Column::UserId.eq(user_id)) .all(&self.orm) .await? .into_iter() .map(Sub::from) .collect()) } /// Everyone's settings for one feed. The scanner merges these into what it fetches /// and downloads, since one file serves the lot. /// In a group, whatever someone has not set on the feed itself comes from their /// subscription to the group, as the group's settings dialog has always said it does. pub async fn subscribers(&self, feed_id: &str, group: Option<&str>) -> Result> { let rows = self .rows( "SELECT coalesce(c.keywords, p.keywords) AS keywords, coalesce(c.auto_download, p.auto_download) AS auto_download, coalesce(c.allow_explicit, p.allow_explicit) AS allow_explicit, coalesce(c.max_new_per_check, p.max_new_per_check) AS max_new_per_check FROM subscriptions c LEFT JOIN subscriptions p ON p.user_id = c.user_id AND p.feed_id = $2 WHERE c.feed_id = $1", vec![feed_id.into(), group.map(str::to_owned).into()], ) .await?; rows.iter() .map(|r| { Ok(Sub { feed_id: feed_id.to_owned(), keywords: keywords(r.try_get("", "keywords")?), auto_download: r.try_get("", "auto_download")?, allow_explicit: r.try_get("", "allow_explicit")?, max_new_per_check: r.try_get("", "max_new_per_check")?, }) }) .collect() } /// Subscribers per feed, for the whole catalogue in one query -- the feed list would /// otherwise ask once per feed. pub async fn subscriber_counts(&self) -> Result> { self.rows("SELECT feed_id, count(*) AS n FROM subscriptions GROUP BY feed_id", vec![]) .await? .iter() .map(|r| Ok((r.try_get("", "feed_id")?, r.try_get("", "n")?))) .collect() } /// Feeds with any audio or video enclosure: the Directory's Podcasts, with the rest Blogs. /// Reaped files keep their rows, so a show whose files have all been purged still counts. pub async fn media_feeds(&self) -> Result> { self.rows( "SELECT DISTINCT feed_id FROM enclosures WHERE mime LIKE 'audio/%' OR mime LIKE 'video/%'", vec![], ) .await? .iter() .map(|r| Ok(r.try_get("", "feed_id")?)) .collect() } /// Who else would miss this file: subscribers other than `user_id` who have starred /// the item or have not read it yet. Deleting is deleting their copy too. pub async fn others_wanting(&self, enclosure_id: i64, user_id: i64) -> Result<(i64, i64)> { // CASE WHEN on the column itself, not `= 1`: a boolean on Postgres, 0 or 1 on SQLite, // and NULL, for someone who never opened the item, falls to the ELSE either way. let r = self .rows( "SELECT sum(CASE WHEN st.flagged THEN 1 ELSE 0 END) AS starred, sum(CASE WHEN st.read THEN 0 ELSE 1 END) AS unread FROM enclosures e JOIN subscriptions s ON s.feed_id = e.feed_id AND s.user_id <> $2 LEFT JOIN entry_state st ON st.user_id = s.user_id AND st.feed_id = e.feed_id AND st.guid = e.guid WHERE e.id = $1", vec![enclosure_id.into(), user_id.into()], ) .await?; let r = r.first().context("an aggregate always returns a row")?; Ok(( r.try_get::>("", "starred")?.unwrap_or(0), r.try_get::>("", "unread")?.unwrap_or(0), )) } pub async fn subscribe(&self, user_id: i64, feed_id: &str) -> Result<()> { self.exec( "INSERT INTO subscriptions (user_id, feed_id) VALUES ($1, $2) ON CONFLICT DO NOTHING", vec![user_id.into(), feed_id.into()], ) .await?; Ok(()) } /// The feeds this person pinned to the top of their list. Kept apart from `Sub`, which is /// what the scanner merges into its policy, and which a pin has nothing to do with. pub async fn pinned_feeds(&self, user_id: i64) -> Result> { Ok(subscriptions::Entity::find() .filter(subscriptions::Column::UserId.eq(user_id)) .filter(subscriptions::Column::Pinned.eq(true)) .all(&self.orm) .await? .into_iter() .map(|s| s.feed_id) .collect()) } /// False when they do not subscribe to it, since there is then no row in their list to pin. pub async fn set_pinned(&self, user_id: i64, feed_id: &str, on: bool) -> Result { let r = subscriptions::Entity::update_many() .col_expr(subscriptions::Column::Pinned, Expr::val(on).into()) .filter(subscriptions::Column::UserId.eq(user_id)) .filter(subscriptions::Column::FeedId.eq(feed_id)) .exec(&self.orm) .await?; Ok(r.rows_affected > 0) } pub async fn unsubscribe(&self, user_id: i64, feed_id: &str) -> Result<()> { subscriptions::Entity::delete_by_id((user_id, feed_id.to_owned())).exec(&self.orm).await?; Ok(()) } /// Overwrites one person's settings for a feed. A None field means: follow the feed. /// Names its columns, so the pin, which is not a setting, is left as it was. pub async fn set_subscription(&self, user_id: i64, sub: &Sub) -> Result<()> { let kw = sub.keywords.as_ref().map(serde_json::to_string).transpose()?; self.exec( "INSERT INTO subscriptions (user_id, feed_id, keywords, auto_download, allow_explicit, max_new_per_check) VALUES ($1, $2, $3, $4, $5, $6) ON CONFLICT (user_id, feed_id) DO UPDATE SET keywords = excluded.keywords, auto_download = excluded.auto_download, allow_explicit = excluded.allow_explicit, max_new_per_check = excluded.max_new_per_check", vec![ user_id.into(), sub.feed_id.clone().into(), kw.into(), sub.auto_download.into(), sub.allow_explicit.into(), sub.max_new_per_check.into(), ], ) .await?; Ok(()) } // ---- users and sessions ---- pub async fn create_user(&self, name: &str, pass_hash: Option<&str>, admin: bool) -> Result { let m = users::ActiveModel { name: Set(name.to_owned()), pass_hash: Set(pass_hash.map(str::to_owned)), is_admin: Set(admin), created: Set(Some(now())), ..Default::default() } .insert(&self.orm) .await?; Ok(m.id) } /// Without regard to case, on either database: lower() both sides, which the unique index /// on lower(name) serves. SQLite's COLLATE NOCASE on the column did this before, and /// Postgres has no such thing. pub async fn user_by_name(&self, name: &str) -> Result> { use sea_orm::sea_query::ExprTrait; Ok(users::Entity::find() .filter(Expr::expr(Func::lower(Expr::col(users::Column::Name))).eq(Func::lower(name))) .one(&self.orm) .await? .map(User::from)) } pub async fn user_by_id(&self, id: i64) -> Result> { Ok(users::Entity::find_by_id(id).one(&self.orm).await?.map(User::from)) } pub async fn users(&self) -> Result> { Ok(users::Entity::find() .order_by_asc(users::Column::Name) .all(&self.orm) .await? .into_iter() .map(User::from) .collect()) } /// Sets some of one person's columns, whatever else is in their row. async fn update_user(&self, id: i64, m: users::ActiveModel) -> Result<()> { users::Entity::update_many() .set(m) .filter(users::Column::Id.eq(id)) .exec(&self.orm) .await?; Ok(()) } pub async fn set_password(&self, id: i64, hash: &str) -> Result<()> { self.update_user(id, users::ActiveModel { pass_hash: Set(Some(hash.to_owned())), ..Default::default() }) .await } pub async fn set_admin(&self, id: i64, admin: bool) -> Result<()> { self.update_user(id, users::ActiveModel { is_admin: Set(admin), ..Default::default() }).await } /// The proxy signs people in by the name it vouches for, so an account made before the proxy /// was set up has to take that name to be found by it. The name is unique, so a taken one is /// refused here as well as by the caller. pub async fn rename_user(&self, id: i64, name: &str) -> Result<()> { self.update_user(id, users::ActiveModel { name: Set(name.to_owned()), ..Default::default() }).await } /// Records a sign-in, to the hour: the proxy vouches for every request, and writing each one /// would buy nothing. pub async fn signed_in(&self, id: i64) -> Result<()> { // Here only: it is implemented for every type, and file-wide it shadows i64::max. use sea_orm::sea_query::ExprTrait; let now = now(); users::Entity::update_many() .col_expr(users::Column::LastLogin, Expr::val(now).into()) .filter(users::Column::Id.eq(id)) .filter(Expr::expr(Func::coalesce([Expr::col(users::Column::LastLogin), Expr::val(0)])).lte(now - 3600)) .exec(&self.orm) .await?; Ok(()) } /// Sessions go with the user: a deleted account must not leave a usable cookie behind. /// The foreign key would take them anyway; this does not rely on it being switched on. pub async fn delete_user(&self, id: i64) -> Result<()> { sessions::Entity::delete_many().filter(sessions::Column::UserId.eq(id)).exec(&self.orm).await?; users::Entity::delete_by_id(id).exec(&self.orm).await?; Ok(()) } pub async fn create_session(&self, user_id: i64, token: &str) -> Result<()> { sessions::ActiveModel { token: Set(token.to_owned()), user_id: Set(user_id), seen: Set(now()) } .insert(&self.orm) .await?; Ok(()) } /// The theme this person chose, and light, dark or auto; None for either until they choose. pub async fn theme(&self, user_id: i64) -> Result<(Option, Option)> { Ok(users::Entity::find_by_id(user_id) .one(&self.orm) .await? .map(|u| (u.theme, u.theme_mode)) .unwrap_or_default()) } pub async fn set_theme(&self, user_id: i64, theme: &str, mode: &str) -> Result<()> { self.update_user( user_id, users::ActiveModel { theme: Set(Some(theme.to_owned())), theme_mode: Set(Some(mode.to_owned())), ..Default::default() }, ) .await } /// The user behind a session cookie, if it is still live. Idle sessions expire after /// `max_idle_secs`; touching `seen` is what keeps a session in daily use alive. pub async fn session_user(&self, token: &str, max_idle_secs: i64) -> Result> { // Here only: it is implemented for every type, and file-wide it shadows i64::max. use sea_orm::sea_query::ExprTrait; let cutoff = now() - max_idle_secs; let found = sessions::Entity::find_by_id(token.to_owned()) .filter(sessions::Column::Seen.gte(cutoff)) .find_also_related(users::Entity) .one(&self.orm) .await? .and_then(|(_, u)| u); if found.is_some() { sessions::Entity::update_many() .col_expr(sessions::Column::Seen, Expr::val(now()).into()) .filter(sessions::Column::Token.eq(token)) .exec(&self.orm) .await?; } else { // Either unknown or timed out; either way it is dead weight. sessions::Entity::delete_many() .filter(sessions::Column::Token.eq(token).or(sessions::Column::Seen.lt(cutoff))) .exec(&self.orm) .await?; } Ok(found.map(User::from)) } pub async fn delete_session(&self, token: &str) -> Result<()> { sessions::Entity::delete_by_id(token.to_owned()).exec(&self.orm).await?; Ok(()) } /// Marks every entry of the given feeds read. Takes a list because an OPML subscription /// holds no entries itself -- marking it read means the feeds inside it. pub async fn mark_all_read(&self, user_id: i64, feed_ids: &[String]) -> Result { let mut n = 0; for id in feed_ids { n += self .exec( "INSERT INTO entry_state (user_id, feed_id, guid, read) SELECT $1, e.feed_id, e.guid, true FROM entries e LEFT JOIN entry_state s ON s.user_id = $1 AND s.feed_id = e.feed_id AND s.guid = e.guid WHERE e.feed_id = $2 AND NOT coalesce(s.read, false) ON CONFLICT (user_id, feed_id, guid) DO UPDATE SET read = true", vec![user_id.into(), id.clone().into()], ) .await? as usize; } Ok(n) } /// The next N enclosures with no file, newest entry first -- what "download latest" /// queues up. pub async fn undownloaded(&self, feed_id: &str, limit: i64) -> Result> { self.rows( "SELECT x.id FROM enclosures x JOIN entries e ON e.feed_id = x.feed_id AND e.guid = x.guid WHERE x.feed_id = $1 AND x.path IS NULL AND x.state <> 'reaped' ORDER BY coalesce(e.published, e.first_seen) DESC LIMIT $2", vec![feed_id.into(), limit.into()], ) .await? .iter() .map(|r| Ok(r.try_get("", "id")?)) .collect() } pub async fn enclosure(&self, id: i64) -> Result> { Ok(enclosures::Entity::find_by_id(id).one(&self.orm).await?.map(EncRow::from)) } /// Read and kept, per person. The row is created on first touch. pub async fn set_entry_flag( &self, user_id: i64, feed_id: &str, guid: &str, field: EntryFlag, on: bool, ) -> Result<()> { let col = match field { EntryFlag::Read => "read", EntryFlag::Flagged => "flagged", }; self.exec( &format!( "INSERT INTO entry_state (user_id, feed_id, guid, {col}) VALUES ($1, $2, $3, $4) ON CONFLICT (user_id, feed_id, guid) DO UPDATE SET {col} = excluded.{col}" ), vec![user_id.into(), feed_id.into(), guid.into(), on.into()], ) .await?; Ok(()) } /// How many files this feed has on disk. Decides whether a feed dropped from an OPML /// can be removed or must be kept. pub async fn downloaded_count(&self, feed_id: &str) -> Result { Ok(enclosures::Entity::find() .filter(enclosures::Column::FeedId.eq(feed_id)) .filter(enclosures::Column::Path.is_not_null()) .count(&self.orm) .await? as i64) } /// Names a feed without touching its conditional-GET validators. An OPML subscription /// takes its name from the document's own . 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 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, true, false) ON CONFLICT (id) DO UPDATE SET url = excluded.url, title = coalesce(feeds.title, excluded.title), group_id = excluded.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 async fn managed_feeds(&self) -> Result<Vec<Managed>> { self.rows( "SELECT id, url, title, group_id FROM feeds 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 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(()) } /// Folds enclosures of one item that `key` says are the same file into the first of them, for /// 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 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 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 LIKE '%?\_=%' ESCAPE '\' OR url LIKE '%&\_=%' ESCAPE '\') ORDER BY id", )) .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 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())); } Entry::Occupied(mut o) => { let (keep, has) = o.get_mut(); if let Some(p) = path { if *has { spare.push(p); } else { // The only copy is the repeat's: Rands' episode 97 was downloaded // under its ?_=2 URL alone. 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", vec![(*keep).into(), id.into()], )) .await?; *has = true; } } tx.execute_raw(Statement::from_sql_and_values( backend, "DELETE FROM enclosures WHERE id = $1", vec![id.into()], )) .await?; gone += 1; } } } tx.commit().await?; Ok((gone, spare)) } /// Stops treating a feed as derived, because it now has its own config entry. 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 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 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 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 backend = self.orm.get_database_backend(); let tx = self.orm.begin().await?; for &(guid, url) in listed { 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. 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().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 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'", vec![feed_id.into()], ) .await? .iter() .map(|r| Ok((r.try_get("", "url")?, r.try_get("", "last_error")?))) .collect() } 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(()) } /// Nothing can be in flight the moment the daemon starts, so any row still marked /// `downloading` is a leftover from a restart or a crash. Left alone it would sit /// there forever: the pending queue skips it and nothing else ever revisits it. pub async fn requeue_interrupted(&self) -> Result<usize> { Ok(self .exec("UPDATE enclosures SET state = 'pending' WHERE state = 'downloading' AND path IS NULL", vec![]) .await? as usize) } /// Puts an enclosure back in the queue so the next scan picks it up. This is how a /// `skipped` verdict (from a filter that has since been changed) gets revisited. pub async fn requeue(&self, id: i64) -> Result<()> { self.exec( "UPDATE enclosures SET state = 'pending', last_error = NULL WHERE id = $1 AND path IS NULL", vec![id.into()], ) .await?; Ok(()) } } #[derive(Debug, Clone, Copy)] pub enum EntryFlag { Read, Flagged, } /// Unix seconds. Everything time-shaped in the DB is stored this way. pub fn now() -> i64 { std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .map(|d| d.as_secs() as i64) .unwrap_or(0) } #[cfg(test)] mod tests { use super::*; #[tokio::test] async fn a_file_wordpress_listed_twice_is_folded_into_one() { let db = Db::memory().await.unwrap(); db.exec_for_test( "INSERT INTO enclosures (id, feed_id, guid, url, path, state) VALUES (1,'f','a','https://x/a.mp3','/d/a-2.mp3','done'), (2,'f','a','https://x/a.mp3?_=2','/d/a.mp3','done'), (3,'f','b','https://x/b.mp3',NULL,'reaped'), (4,'f','b','https://x/b.mp3?_=2','/d/b.mp3','done'), (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).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] async fn every_sort_column_runs_and_orders_both_ways() { let db = Db::memory().await.unwrap(); db.exec_for_test( "INSERT INTO users (id, name, is_admin) VALUES (1,'ray',true); INSERT INTO subscriptions (user_id, feed_id) VALUES (1,'f'),(1,'g'); INSERT INTO feeds (id, url, title) VALUES ('f','u','Zebra'),('g','v','Aardvark'); INSERT INTO entries (feed_id, guid, title, first_seen) VALUES ('f','a','banana',100),('g','b','Apple',200),('f','c','cherry',300); 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)).await .unwrap() .into_iter() .map(|e| e.guid) .collect() }; assert_eq!(order("title", "asc").await, ["b", "a", "c"], "Apple, banana, cherry: case folded"); assert_eq!(order("title", "desc").await, ["c", "a", "b"]); assert_eq!(order("feed", "asc").await, ["b", "c", "a"], "Aardvark, then Zebra's newest first"); assert_eq!(order("type", "asc").await, ["a", "b", "c"], "audio, image, video"); assert_eq!(order("size", "desc").await, ["c", "a", "b"]); // An item with no file has no size or type: last going down, first going up, on both // databases (they disagree about NULL unless told). db.exec_for_test("INSERT INTO entries (feed_id, guid, title, first_seen) VALUES ('f','n','no file',50);") .await .unwrap(); assert_eq!(order("size", "desc").await.last().map(String::as_str), Some("n")); assert_eq!(order("type", "asc").await.first().map(String::as_str), Some("n")); db.exec_for_test("DELETE FROM entries WHERE guid = 'n';").await.unwrap(); assert_eq!(order("published", "desc").await, ["c", "b", "a"]); db.set_entry_flag(1, "f", "a", EntryFlag::Flagged, true).await.unwrap(); assert_eq!(order("kept", "desc").await[0], "a"); // A pin does not lift an item above the others in any other sort (#36). assert_eq!(order("published", "desc").await, ["c", "b", "a"]); // An unknown column or direction is newest first; the name itself never reaches the SQL. assert_eq!(order("title; DROP TABLE entries", "sideways").await, ["c", "b", "a"]); assert!(!order_sql("x'; --", "asc").contains("x'")); } #[tokio::test] async fn deleting_a_shared_file_asks_about_everyone_else() { let db = Db::memory().await.unwrap(); db.exec_for_test( "INSERT INTO users (id, name, is_admin) VALUES (1,'ray',true),(2,'sam',false),(3,'kit',false); INSERT INTO subscriptions (user_id, feed_id) VALUES (1,'f'),(2,'f'),(3,'f'); 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. assert_eq!(db.others_wanting(1, 1).await.unwrap(), (0, 2)); // Sam reads it, Kit stars it. db.set_entry_flag(2, "f", "a", EntryFlag::Read, true).await.unwrap(); db.set_entry_flag(3, "f", "a", EntryFlag::Flagged, true).await.unwrap(); assert_eq!(db.others_wanting(1, 1).await.unwrap(), (1, 1), "one starred it, one has not played it"); // Asking as Kit, only Ray and Sam count -- and Kit's own star is not a reason to // warn Kit. db.set_entry_flag(1, "f", "a", EntryFlag::Read, true).await.unwrap(); assert_eq!(db.others_wanting(1, 3).await.unwrap(), (0, 0)); } #[tokio::test] async fn read_state_belongs_to_one_person() { let db = Db::memory().await.unwrap(); db.exec_for_test( "INSERT INTO users (id, name, is_admin) VALUES (1,'ray',true),(2,'sam',false); 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); assert_eq!(db.unread_count(2, "f").await.unwrap(), 2); db.set_entry_flag(1, "f", "a", EntryFlag::Read, true).await.unwrap(); assert_eq!(db.unread_count(1, "f").await.unwrap(), 1, "ray read one of them"); assert_eq!(db.unread_count(2, "f").await.unwrap(), 2, "sam has read nothing"); // Starring and position are just as private. db.set_entry_flag(1, "f", "b", EntryFlag::Flagged, true).await.unwrap(); db.set_position(2, "f", "b", 42, Some(600)).await.unwrap(); let order = order_sql("published", "desc"); let page = async |user| db.entries_in(user, Some("f"), Filter::All, None, 0, 50, &order).await.unwrap(); let (ray, sam) = (page(1).await, page(2).await); let ray_b = ray.iter().find(|e| e.guid == "b").unwrap(); let sam_b = sam.iter().find(|e| e.guid == "b").unwrap(); assert!(ray_b.flagged && ray_b.position == 0); assert!(!sam_b.flagged && sam_b.position == 42); // So is the length sam's player measured. assert_ne!(ray_b.duration, Some(600)); assert_eq!(sam_b.duration, Some(600)); // Marking a whole feed read is likewise one person's business. assert_eq!(db.mark_all_read(2, &["f".to_string()]).await.unwrap(), 2); assert_eq!(db.unread_count(2, "f").await.unwrap(), 0); assert_eq!(db.unread_count(1, "f").await.unwrap(), 1); } #[tokio::test] async fn schema_is_idempotent_and_summary_handles_unknown_feeds() { let db = Db::memory().await.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").await.unwrap(); assert_eq!(sum.last_checked, None); assert_eq!(sum.entries, 0); assert_eq!(sum.downloaded, 0); } #[tokio::test] async fn a_renamed_account_keeps_everything_but_its_name() { let db = Db::memory().await.unwrap(); let ray = db.create_user("rays", None, true).await.unwrap(); db.create_user("sam", None, false).await.unwrap(); db.subscribe(ray, "f").await.unwrap(); db.rename_user(ray, "rays@sdf1.net").await.unwrap(); assert!(db.user_by_name("rays").await.unwrap().is_none()); let renamed = db.user_by_name("RAYS@sdf1.net").await.unwrap().unwrap(); assert_eq!((renamed.id, renamed.is_admin), (ray, true), "same account, still the admin"); assert_eq!(db.subscriptions_for(ray).await.unwrap().len(), 1, "and still subscribed"); assert!(db.rename_user(ray, "sam").await.is_err(), "a taken name is refused"); } #[tokio::test] async fn an_account_knows_when_it_was_made_and_last_signed_in() { let db = Db::memory().await.unwrap(); let id = db.create_user("ray", None, true).await.unwrap(); let get = async || db.user_by_id(id).await.unwrap().unwrap(); assert!(get().await.created.is_some_and(|t| t > 0)); assert_eq!(get().await.last_login, None, "made, but never signed in"); 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)).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)).await.unwrap(); db.signed_in(id).await.unwrap(); assert!(get().await.last_login.unwrap() >= first); } #[tokio::test] async fn the_first_admin_starts_with_the_catalogue_and_only_once() { // 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',true);").await .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().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().await, 1); } #[tokio::test] async fn every_filter_works_with_and_without_a_search_term() { // Regression: the search clause used to be omitted when no term was given, while // ?2 was still bound -- rusqlite rejects a parameter the statement never mentions, // so plain filtering failed with "Wrong number of parameters passed to query". let db = Db::memory().await.unwrap(); db.exec_for_test( "INSERT INTO entries (feed_id, guid, title, description, first_seen, duration) VALUES ('f','a','Alpha dive','notes one', 100,NULL), ('f','b','Beta', 'notes two', 200,NULL), ('f','c','Gamma dive','notes three',300,NULL), ('f','d','Delta', 'notes four', 400,NULL), ('f','e','Epsilon', 'notes five', 500,45), ('f','g','Gimel', 'notes six', 600,NULL), ('f','h','Heth', 'notes seven',700,900); INSERT INTO enclosures (id, feed_id, guid, url, path, state) VALUES (1,'f','b','u1','/tmp/b','done'); -- Read and starred belong to a person now, so say which one. INSERT INTO users (id, name, is_admin) VALUES (7,'reader',true); INSERT INTO entry_state (user_id, feed_id, guid, read, flagged, position) VALUES (7,'f','b',true,false,0), (7,'f','c',true,true,0), -- Started, length unknown: this is Currently Listening. (7,'f','d',false,false,42), -- 42 of 45 seconds is past the 90% the player calls finished. (7,'f','e',true,false,42), -- Barely touched (opened, closed within seconds): not Currently Listening. (7,'f','g',false,false,3), -- Opened, and so read, but 42 of 900 seconds in: still Currently Listening. -- Filtering on read hid exactly these (issue #14). (7,'f','h',true,false,42);", ).await .unwrap(); for f in [Filter::All, Filter::Unread, Filter::Downloaded, Filter::Flagged, Filter::InProgress] { // Both paths must run without erroring, and agree with each other. let order = order_sql("published", "desc"); let rows = db.entries_in(7, Some("f"), f, None, 0, 50, &order).await.unwrap(); let n = db.count_in(7, Some("f"), f, None).await.unwrap(); assert_eq!(rows.len() as i64, n, "{f:?} count disagrees with the page"); let rows = db.entries_in(7, Some("f"), f, Some("dive"), 0, 50, &order).await.unwrap(); let n = db.count_in(7, Some("f"), f, Some("dive")).await.unwrap(); assert_eq!(rows.len() as i64, n, "{f:?} with search disagrees"); } assert_eq!(db.count_in(7, Some("f"), Filter::All, None).await.unwrap(), 7); assert_eq!(db.count_in(7, Some("f"), Filter::Unread, None).await.unwrap(), 3); assert_eq!(db.count_in(7, Some("f"), Filter::Downloaded, None).await.unwrap(), 1); assert_eq!(db.count_in(7, Some("f"), Filter::Flagged, None).await.unwrap(), 1); assert_eq!(db.count_in(7, Some("f"), Filter::All, Some("dive")).await.unwrap(), 2); assert_eq!(db.count_in(7, Some("f"), Filter::All, Some("NOTES two")).await.unwrap(), 1, "search is case-insensitive and covers the description"); // Currently Listening: started, not finished, and not just an accidental tap. let order = order_sql("published", "desc"); let listening = async || { let rows = db.entries_in(7, Some("f"), Filter::InProgress, None, 0, 50, &order).await.unwrap(); rows.into_iter().map(|e| e.guid).collect::<Vec<_>>() }; assert_eq!(listening().await, ["h", "d"]); // The player's measured length is the one that counts, in place of a missing one or over // the feed's: d, 42 of a measured 45 seconds, is finished; e, which its feed calls 45 // seconds long, is 42 into a 9000-second file and is not. Times left use it too. db.set_position(7, "f", "d", 42, Some(45)).await.unwrap(); db.set_position(7, "f", "e", 42, Some(9000)).await.unwrap(); assert_eq!(listening().await, ["h", "e"]); let rows = db.entries_in(7, Some("f"), Filter::All, None, 0, 50, &order).await.unwrap(); assert_eq!(rows.iter().find(|r| r.guid == "e").unwrap().duration, Some(9000)); // A save without one (before the player knows) keeps the length already measured. db.set_position(7, "f", "e", 43, None).await.unwrap(); assert_eq!(listening().await, ["h", "e"]); } #[tokio::test] async fn pending_takes_the_latest_episodes_first() { // A cap of 3 must mean the three newest, not the three recorded first. let db = Db::memory().await.unwrap(); db.exec_for_test( "INSERT INTO entries (feed_id, guid, published, first_seen) VALUES ('f','old',100,100), ('f','mid',200,200), ('f','new',300,300); INSERT INTO enclosures (id, feed_id, guid, url, state) VALUES (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"); } #[tokio::test] async fn a_restart_requeues_interrupted_downloads() { let db = Db::memory().await.unwrap(); db.exec_for_test( "INSERT INTO enclosures (id, feed_id, guid, url, state, path) VALUES (1,'f','a','u1','downloading',NULL), (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 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] async fn a_show_takes_over_what_its_creator_held() { let db = Db::memory().await.unwrap(); db.exec_for_test( "INSERT INTO users (id, name, is_admin) VALUES (1,'ray',true); INSERT INTO enclosures (id, feed_id, guid, url, state, path, last_error) VALUES (1,'creator','a','u1','done','/x/a.mp3',NULL), (2,'creator','b','u2','skipped',NULL,'explicit'), (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',true);", ).await .unwrap(); 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").await.unwrap(); assert_eq!(skipped.get("u2").map(String::as_str), Some("explicit")); assert_eq!(skipped.len(), 1); assert!(db.skipped_by_filter("other").await.unwrap().is_empty(), "torrents disabled is not a filter"); } #[tokio::test] async fn a_feed_in_a_group_follows_your_settings_on_the_group() { let db = Db::memory().await.unwrap(); db.exec_for_test( "INSERT INTO users (id, name, is_admin) VALUES (1,'ray',true),(2,'sam',false); INSERT INTO subscriptions (user_id, feed_id, allow_explicit) VALUES (1,'group',true),(1,'show',NULL),(2,'group',true),(2,'show',false);", ).await .unwrap(); let explicit = async |group| -> Vec<Option<bool>> { let mut v: Vec<_> = db.subscribers("show", group).await.unwrap().into_iter().map(|s| s.allow_explicit).collect(); v.sort(); v }; assert_eq!(explicit(Some("group")).await, [Some(false), Some(true)], "ray inherits; sam's own choice on the show wins"); assert_eq!(explicit(None).await, [None, Some(false)], "outside a group nothing is inherited"); } #[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").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'").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").await.unwrap(); assert_eq!(db.feed_summary("f").await.unwrap().error_since, Some(first)); 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"); } #[tokio::test] async fn a_pin_survives_saving_the_feeds_settings_and_needs_a_subscription() { let db = Db::memory().await.unwrap(); let me = db.create_user("pat", None, false).await.unwrap(); assert!(!db.set_pinned(me, "f", true).await.unwrap(), "not subscribed: nothing to pin"); db.subscribe(me, "f").await.unwrap(); assert!(db.set_pinned(me, "f", true).await.unwrap()); // set_subscription writes the rest of the row; it must leave the pin alone. db.set_subscription(me, &Sub { feed_id: "f".into(), auto_download: Some(false), ..Default::default() }).await .unwrap(); assert!(db.pinned_feeds(me).await.unwrap().contains("f")); db.set_pinned(me, "f", false).await.unwrap(); assert!(db.pinned_feeds(me).await.unwrap().is_empty()); } #[test] fn postgres_is_asked_for_no_notices_and_a_password_never_reaches_a_log() { assert_eq!(url_for("postgres://u:p@h/d"), format!("postgres://u:p@h/d?{QUIET}")); assert_eq!(url_for("postgres://u:p@h/d?sslmode=disable"), format!("postgres://u:p@h/d?sslmode=disable&{QUIET}")); assert_eq!(url_for("postgres://u:p@h/d?options=-c%20x%3Dy"), "postgres://u:p@h/d?options=-c%20x%3Dy", "theirs stands"); assert_eq!(url_for("/data/state.db"), "sqlite:///data/state.db?mode=rwc"); assert_eq!(redact("postgres://ipodderx:s3cret@h:5433/ipodderx"), "postgres://ipodderx:***@h:5433/ipodderx"); } #[tokio::test] async fn the_configuration_goes_in_once_and_comes_back_as_it_went() { let db = Db::memory().await.unwrap(); assert!(db.stored_config().await.unwrap().is_none(), "nothing until it is taken in"); let mut cfg: crate::config::Config = toml::from_str( r#" [general] schedule = "every 2h" max_total_gb = 1.5 [feeds.a] url = "http://x/a.xml" keywords = ["one", "two"] [feeds.b] url = "http://x/b.xml" password = "secret" "#, ) .unwrap(); assert!(db.import_config(&cfg).await.unwrap()); assert!(!db.import_config(&cfg).await.unwrap(), "a second import finds the first and stands back"); let (stored, feeds) = db.stored_config().await.unwrap().unwrap(); assert_eq!(stored, crate::config::Stored::of(&cfg)); assert_eq!(feeds.keys().collect::<Vec<_>>(), ["a", "b"]); assert_eq!(feeds["a"].keywords, ["one", "two"]); assert_eq!(feeds["b"].password.as_deref(), Some("secret")); // Storing is the whole catalogue: a feed taken out goes, one put in arrives. cfg.feeds.remove("a"); cfg.feeds.insert("c".into(), cfg.feeds["b"].clone()); cfg.general.schedule = "every 30m".into(); db.store_config(&cfg).await.unwrap(); let (stored, feeds) = db.stored_config().await.unwrap().unwrap(); assert_eq!(stored.schedule, "every 30m"); assert_eq!(feeds.keys().collect::<Vec<_>>(), ["b", "c"]); } #[tokio::test] async fn enclosure_url_is_the_dedupe_key() { let db = Db::memory().await.unwrap(); let insert = "INSERT INTO enclosures (feed_id, guid, url, state) VALUES ('f', 'g', 'http://x/a.mp3', 'pending')"; db.exec_for_test(insert).await.unwrap(); assert!(db.exec_for_test(insert).await.is_err(), "duplicate url must be rejected"); } }