//! SQLite state. Replaces the per-feed .ipxd plists, history.dat and qmcache.dat. use anyhow::{Context, Result}; use rusqlite::{Connection, OptionalExtension, params}; 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. pub struct Db { conn: Mutex, } const SCHEMA: &str = " CREATE TABLE IF NOT EXISTS feeds ( id TEXT PRIMARY KEY, url TEXT NOT NULL, title TEXT, image TEXT, etag TEXT, last_modified TEXT, last_checked INTEGER, ttl_mins INTEGER, last_error TEXT, -- 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, read INTEGER NOT NULL DEFAULT 0, flagged INTEGER NOT NULL DEFAULT 0, image TEXT, duration INTEGER, episode INTEGER, season INTEGER, -- Seconds into the audio, so playback resumes where it was left. position INTEGER NOT NULL DEFAULT 0, 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, created INTEGER NOT NULL ); -- 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, created INTEGER NOT NULL, PRIMARY KEY (user_id, feed_id) ); -- Read, starred 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, 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, created INTEGER NOT NULL, 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)] 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, } /// 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, pub orphaned: bool, } /// Adds columns that later versions introduced. CREATE TABLE IF NOT EXISTS does nothing to /// a table that already exists, so an installed database needs them added explicitly. fn migrate(conn: &Connection) -> Result<()> { let wanted: &[(&str, &str, &str)] = &[ ("feeds", "image", "TEXT"), ("feeds", "orphaned", "INTEGER NOT NULL DEFAULT 0"), ("feeds", "group_id", "TEXT"), ("feeds", "managed", "INTEGER NOT NULL DEFAULT 0"), ("entries", "image", "TEXT"), ("entries", "duration", "INTEGER"), ("entries", "episode", "INTEGER"), ("entries", "season", "INTEGER"), ("entries", "position", "INTEGER NOT NULL DEFAULT 0"), ]; for (table, column, ty) in wanted { let mut stmt = conn.prepare(&format!("PRAGMA table_info({table})"))?; let existing: Vec = stmt .query_map([], |r| r.get::<_, String>(1))? .collect::>>()?; if !existing.iter().any(|c| c == column) { tracing::info!(table, column, "adding column"); conn.execute_batch(&format!("ALTER TABLE {table} ADD COLUMN {column} {ty}"))?; } } Ok(()) } /// What `ipx list` shows next to each configured feed. #[derive(Debug, Default)] pub struct FeedSummary { pub title: Option, pub image: 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, pub entries: i64, pub downloaded: i64, } impl Db { pub fn open(path: &Path) -> Result { if let Some(dir) = path.parent() { 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")?; Ok(Self { conn: Mutex::new(conn) }) } /// In-memory database, for tests. #[cfg(test)] pub fn memory() -> Result { let conn = Connection::open_in_memory()?; conn.execute_batch(SCHEMA)?; // Same path as a real open, so a column added only in migrate() cannot pass the // tests while being missing in production (or the reverse). migrate(&conn)?; Ok(Self { conn: Mutex::new(conn) }) } #[cfg(test)] pub fn exec_for_test(&self, sql: &str) -> Result<()> { self.conn.lock().unwrap().execute_batch(sql)?; 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) 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, ..Default::default() }) }, ) .optional()? .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), )?; 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 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()? .unwrap_or_default()) } /// Upsert after a successful poll. Clears any previous error. pub 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>, ) -> Result<()> { let conn = self.conn.lock().unwrap(); conn.execute( "INSERT INTO feeds (id, url, title, etag, last_modified, last_checked, ttl_mins, last_error, image) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, NULL, ?8) 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), last_error = NULL", rusqlite::params![feed_id, url, title, etag, last_modified, now(), ttl_mins.map(|t| t as i64), image], )?; 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, last_error = NULL", rusqlite::params![feed_id, url, now()], )?; 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], )?; Ok(()) } pub fn set_feed_error(&self, feed_id: &str, url: &str, msg: &str) -> Result<()> { let conn = self.conn.lock().unwrap(); conn.execute( "INSERT INTO feeds (id, url, last_checked, last_error) VALUES (?1, ?2, ?3, ?4) ON CONFLICT(id) DO UPDATE SET last_checked = excluded.last_checked, last_error = excluded.last_error", rusqlite::params![feed_id, url, now(), msg], )?; Ok(()) } /// Returns true when this entry had not been seen before. /// /// A changed description or title flips `read` back to 0, which is what the original's /// textDiff dance was ultimately for -- minus the diff markup, which the UI can do. 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, read, flagged, image, duration, episode, season) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, 0, 0, ?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 { // The SET expressions see the pre-update row, so this compares old vs new. 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), read = CASE WHEN description IS NOT ?4 OR title IS NOT ?3 THEN 0 ELSE read END 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 ], )?; } 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 fn mark_downloaded(&self, url: &str, path: &std::path::Path, bytes: u64) -> Result<()> { let conn = self.conn.lock().unwrap(); conn.execute( "UPDATE enclosures SET state = 'done', path = ?2, bytes_done = ?3, downloaded_at = ?4, last_error = NULL WHERE url = ?1", rusqlite::params![url, path.to_string_lossy(), bytes as i64, now()], )?; 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 fn mark_enclosure(&self, url: &str, state: &str, error: Option<&str>) -> Result<()> { let conn = self.conn.lock().unwrap(); conn.execute( "UPDATE enclosures SET state = ?2, last_error = ?3 WHERE url = ?1", rusqlite::params![url, state, error], )?; Ok(()) } pub fn record_enclosure( &self, feed_id: &str, guid: &str, enc: &crate::feed::Enclosure, ) -> Result { let conn = self.conn.lock().unwrap(); let inserted = conn.execute( "INSERT OR IGNORE INTO enclosures (feed_id, guid, url, mime, length, state) VALUES (?1, ?2, ?3, ?4, ?5, 'pending')", rusqlite::params![feed_id, guid, enc.url, enc.mime, enc.length], )?; 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 fn pending(&self, feed_id: &str, limit: usize) -> Result> { let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare( // Newest first: a cap of 3 should mean the three latest episodes, not the // three that happen to have been recorded first. "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", )?; let rows = stmt .query_map(rusqlite::params![feed_id, limit as i64], |r| { Ok(Pending { id: r.get(0)?, url: r.get(1)?, mime: r.get(2)? }) })? .collect::>>()?; Ok(rows) } } /// 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 fn reap_candidates(&self) -> Result> { let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare( "SELECT e.id, e.url, e.path, e.bytes_done, coalesce(e.downloaded_at, 0), CASE WHEN coalesce(readers.n, 0) >= coalesce(subs.n, 0) THEN 1 ELSE 0 END FROM enclosures e LEFT JOIN (SELECT feed_id, count(*) n FROM subscriptions GROUP BY feed_id) subs ON subs.feed_id = e.feed_id LEFT JOIN (SELECT feed_id, guid, count(*) n FROM entry_state WHERE read = 1 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 = 1) ORDER BY 6 DESC, coalesce(e.downloaded_at, 0) ASC, e.id ASC", )?; let rows = stmt .query_map([], |r| { Ok(Candidate { id: r.get(0)?, url: r.get(1)?, path: r.get(2)?, bytes: r.get(3)?, age_key: r.get(4)?, read: r.get::<_, i64>(5)? != 0, }) })? .collect::>>()?; Ok(rows) } /// The row survives the file: that is what stops a reaped episode being re-downloaded. pub fn mark_reaped(&self, id: i64) -> Result<()> { let conn = self.conn.lock().unwrap(); conn.execute( "UPDATE enclosures SET state = 'reaped', path = NULL WHERE id = ?1", [id], )?; Ok(()) } /// Rows claiming a file that is no longer there (someone deleted it by hand). pub fn missing_files(&self) -> Result> { let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare("SELECT id, path FROM enclosures WHERE path IS NOT NULL")?; let rows = stmt .query_map([], |r| Ok((r.get(0)?, r.get::<_, String>(1)?)))? .collect::>>()?; Ok(rows .into_iter() .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 fn prune_entries(&self, older_than: i64) -> Result { let conn = self.conn.lock().unwrap(); let n = conn.execute( "DELETE FROM entries WHERE flagged = 0 AND 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)", [older_than], )?; Ok(n) } } impl Db { pub fn unread_count(&self, user_id: i64, feed_id: &str) -> Result { let conn = self.conn.lock().unwrap(); Ok(conn.query_row( "SELECT count(*) 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 coalesce(s.read, 0) = 0", rusqlite::params![feed_id, user_id], |r| r.get(0), )?) } /// (pending, downloaded) across all feeds, for the status command. pub fn counts(&self) -> Result<(i64, i64)> { let conn = self.conn.lock().unwrap(); Ok(( conn.query_row("SELECT count(*) FROM enclosures WHERE state = 'pending'", [], |r| r.get(0))?, conn.query_row("SELECT count(*) FROM enclosures WHERE path IS NOT NULL", [], |r| r.get(0))?, )) } } /// 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, } /// The search clause. `?2` is referenced unconditionally -- binding a parameter the /// statement does not mention is an error, so an empty needle short-circuits instead. const SEARCH: &str = "(?2 = '' OR lower(coalesce(e.title, '')) LIKE ?2 OR lower(coalesce(e.description, '')) LIKE ?2)"; /// Which slice of a feed the UI is asking for. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum Filter { All, Unread, Downloaded, Flagged, } impl Filter { pub fn parse(s: &str) -> Self { match s { "unread" => Self::Unread, "downloaded" => Self::Downloaded, "flagged" => Self::Flagged, _ => 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 { Self::All => "1=1", Self::Unread => "coalesce(s.read, 0) = 0", Self::Flagged => "coalesce(s.flagged, 0) = 1", 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)" } } } } #[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 a feed's entries, newest first, each with its enclosures attached. /// `search` matches title and description, case-insensitively. pub fn entries( &self, user_id: i64, feed_id: &str, filter: Filter, search: Option<&str>, offset: i64, limit: i64, ) -> Result> { let conn = self.conn.lock().unwrap(); let like = search .map(|q| format!("%{}%", q.trim().to_lowercase())) .unwrap_or_default(); let sql = format!( "SELECT e.guid, e.feed_id, e.title, e.link, e.published, e.description, coalesce(s.read, 0), coalesce(s.flagged, 0), e.image, e.duration, e.episode, e.season, coalesce(s.position, 0) FROM entries e LEFT JOIN entry_state s ON s.user_id = ?5 AND s.feed_id = e.feed_id AND s.guid = e.guid WHERE e.feed_id = ?1 AND {} AND {SEARCH} ORDER BY coalesce(e.published, e.first_seen) DESC, e.rowid DESC LIMIT ?4 OFFSET ?3", filter.sql() ); let mut stmt = conn.prepare(&sql)?; let map = |r: &rusqlite::Row| -> rusqlite::Result { Ok(EntryRow { guid: r.get(0)?, feed_id: r.get(1)?, title: r.get(2)?, link: r.get(3)?, published: r.get(4)?, description: r.get(5)?, read: r.get::<_, i64>(6)? != 0, flagged: r.get::<_, i64>(7)? != 0, image: r.get(8)?, duration: r.get(9)?, episode: r.get(10)?, season: r.get(11)?, position: r.get(12)?, enclosures: vec![], }) }; let mut rows: Vec = stmt .query_map(rusqlite::params![feed_id, like, offset, limit, user_id], map)? .collect::>>()?; if rows.is_empty() { return Ok(rows); } // Only the guids on this page, so a feed with thousands of entries stays cheap. let placeholders = std::iter::repeat_n("?", rows.len()).collect::>().join(","); let sql = format!( "SELECT id, feed_id, guid, url, mime, length, path, state, last_error FROM enclosures WHERE feed_id = ? AND guid IN ({placeholders}) ORDER BY id" ); let mut params: Vec<&dyn rusqlite::ToSql> = Vec::with_capacity(rows.len() + 1); params.push(&feed_id); for row in &rows { params.push(&row.guid); } let mut stmt = conn.prepare(&sql)?; let encs = stmt .query_map(params.as_slice(), |r| { Ok(EncRow { id: r.get(0)?, feed_id: r.get(1)?, guid: r.get(2)?, url: r.get(3)?, mime: r.get(4)?, length: r.get(5)?, path: r.get(6)?, state: r.get(7)?, last_error: r.get(8)?, }) })? .collect::>>()?; for enc in encs { if let Some(row) = rows.iter_mut().find(|r| r.guid == enc.guid) { row.enclosures.push(enc); } } Ok(rows) } /// How many entries match, so the UI knows whether there is another page. pub fn count_entries( &self, user_id: i64, feed_id: &str, filter: Filter, search: Option<&str>, ) -> Result { let conn = self.conn.lock().unwrap(); let like = search .map(|q| format!("%{}%", q.trim().to_lowercase())) .unwrap_or_default(); let sql = format!( "SELECT count(*) FROM entries e LEFT JOIN entry_state s ON s.user_id = ?3 AND s.feed_id = e.feed_id AND s.guid = e.guid WHERE e.feed_id = ?1 AND {} AND {SEARCH}", filter.sql() ); Ok(conn.query_row(&sql, rusqlite::params![feed_id, like, user_id], |r| r.get(0))?) } /// Where playback got to, so it resumes there next time -- for this listener only. pub fn set_position(&self, user_id: i64, feed_id: &str, guid: &str, secs: i64) -> Result<()> { let conn = self.conn.lock().unwrap(); conn.execute( "INSERT INTO entry_state (user_id, feed_id, guid, position) VALUES (?1, ?2, ?3, ?4) ON CONFLICT(user_id, feed_id, guid) DO UPDATE SET position = excluded.position", rusqlite::params![user_id, feed_id, guid, secs.max(0)], )?; Ok(()) } /// Marks every entry in a feed read, for the "mark all read" button. /// Moves a single-user library onto an account: everything read, starred or part-played /// becomes that person's, and they subscribe to every feed already in the catalogue. /// Runs once -- the moment there is a first account and no subscriptions yet. pub fn adopt_existing_library(&self, user_id: i64, catalogue: &[String]) -> Result { let conn = self.conn.lock().unwrap(); let already: i64 = conn.query_row("SELECT count(*) FROM subscriptions", [], |r| r.get(0))?; if already > 0 { return Ok(0); } let moved = conn.execute( "INSERT INTO entry_state (user_id, feed_id, guid, read, flagged, position) SELECT ?1, feed_id, guid, read, flagged, position FROM entries WHERE read = 1 OR flagged = 1 OR position > 0 ON CONFLICT(user_id, feed_id, guid) DO NOTHING", [user_id], )?; for id in catalogue { conn.execute( "INSERT OR IGNORE INTO subscriptions (user_id, feed_id, created) VALUES (?1, ?2, ?3)", params![user_id, id, now()], )?; } // Feeds that exist only in the database (OPML children) count too. conn.execute( "INSERT OR IGNORE INTO subscriptions (user_id, feed_id, created) SELECT ?1, id, ?2 FROM feeds", params![user_id, now()], )?; Ok(moved) } // ---- subscriptions ---- /// What this person wants from a feed. Absent means they do not subscribe at all. pub fn subscription(&self, user_id: i64, feed_id: &str) -> Result> { let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare( "SELECT keywords, auto_download, allow_explicit, max_new_per_check FROM subscriptions WHERE user_id = ?1 AND feed_id = ?2", )?; let mut rows = stmt.query(params![user_id, feed_id])?; Ok(match rows.next()? { Some(r) => Some(Sub { feed_id: feed_id.to_string(), keywords: r .get::<_, Option>(0)? .and_then(|j| serde_json::from_str(&j).ok()), auto_download: r.get::<_, Option>(1)?.map(|v| v != 0), allow_explicit: r.get::<_, Option>(2)?.map(|v| v != 0), max_new_per_check: r.get(3)?, }), None => None, }) } /// Every feed this person subscribes to, with their settings. pub fn subscriptions_for(&self, user_id: i64) -> Result> { let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare( "SELECT feed_id, keywords, auto_download, allow_explicit, max_new_per_check FROM subscriptions WHERE user_id = ?1", )?; let out = stmt .query_map([user_id], |r| { Ok(Sub { feed_id: r.get(0)?, keywords: r .get::<_, Option>(1)? .and_then(|j| serde_json::from_str(&j).ok()), auto_download: r.get::<_, Option>(2)?.map(|v| v != 0), allow_explicit: r.get::<_, Option>(3)?.map(|v| v != 0), max_new_per_check: r.get(4)?, }) })? .collect::>>()?; Ok(out) } /// Everyone's settings for one feed. The scanner merges these into what it fetches /// and downloads, since one file serves the lot. pub fn subscribers(&self, feed_id: &str) -> Result> { let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare( "SELECT keywords, auto_download, allow_explicit, max_new_per_check FROM subscriptions WHERE feed_id = ?1", )?; let out = stmt .query_map([feed_id], |r| { Ok(Sub { feed_id: feed_id.to_string(), keywords: r .get::<_, Option>(0)? .and_then(|j| serde_json::from_str(&j).ok()), auto_download: r.get::<_, Option>(1)?.map(|v| v != 0), allow_explicit: r.get::<_, Option>(2)?.map(|v| v != 0), max_new_per_check: r.get(3)?, }) })? .collect::>>()?; Ok(out) } /// Subscribers per feed, for the whole catalogue in one query -- the feed list would /// otherwise ask once per feed. pub fn subscriber_counts(&self) -> Result> { let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare("SELECT feed_id, count(*) FROM subscriptions GROUP BY feed_id")?; let out = stmt .query_map([], |r| Ok((r.get::<_, String>(0)?, r.get::<_, i64>(1)?)))? .collect::>>()?; Ok(out) } /// 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 fn others_wanting(&self, enclosure_id: i64, user_id: i64) -> Result<(i64, i64)> { let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare( "SELECT sum(CASE WHEN coalesce(st.flagged, 0) = 1 THEN 1 ELSE 0 END), sum(CASE WHEN coalesce(st.read, 0) = 0 THEN 1 ELSE 0 END) 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", )?; let (starred, unread) = stmt.query_row(params![enclosure_id, user_id], |r| { Ok((r.get::<_, Option>(0)?.unwrap_or(0), r.get::<_, Option>(1)?.unwrap_or(0))) })?; Ok((starred, unread)) } pub fn subscribe(&self, user_id: i64, feed_id: &str) -> Result<()> { let conn = self.conn.lock().unwrap(); conn.execute( "INSERT OR IGNORE INTO subscriptions (user_id, feed_id, created) VALUES (?1, ?2, ?3)", params![user_id, feed_id, now()], )?; Ok(()) } pub fn unsubscribe(&self, user_id: i64, feed_id: &str) -> Result<()> { let conn = self.conn.lock().unwrap(); conn.execute( "DELETE FROM subscriptions WHERE user_id = ?1 AND feed_id = ?2", params![user_id, feed_id], )?; Ok(()) } /// How many people want this feed. Nobody means it stops being scanned. pub fn subscriber_count(&self, feed_id: &str) -> Result { let conn = self.conn.lock().unwrap(); Ok(conn.query_row( "SELECT count(*) FROM subscriptions WHERE feed_id = ?1", [feed_id], |r| r.get(0), )?) } /// Overwrites one person's settings for a feed. A None field means: follow the feed. pub fn set_subscription(&self, user_id: i64, sub: &Sub) -> Result<()> { let conn = self.conn.lock().unwrap(); let kw = sub .keywords .as_ref() .map(|k| serde_json::to_string(k)) .transpose()?; conn.execute( "INSERT INTO subscriptions (user_id, feed_id, keywords, auto_download, allow_explicit, max_new_per_check, created) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7) 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", params![ user_id, sub.feed_id, kw, sub.auto_download.map(|v| v as i64), sub.allow_explicit.map(|v| v as i64), sub.max_new_per_check, now() ], )?; Ok(()) } /// Feeds with at least one subscriber. What the scanner walks. pub fn subscribed_feed_ids(&self) -> Result> { let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare("SELECT DISTINCT feed_id FROM subscriptions")?; let out = stmt .query_map([], |r| r.get::<_, String>(0))? .collect::>>()?; Ok(out) } // ---- users and sessions ---- pub fn create_user(&self, name: &str, pass_hash: Option<&str>, admin: bool) -> Result { let conn = self.conn.lock().unwrap(); conn.execute( "INSERT INTO users (name, pass_hash, is_admin, created) VALUES (?1, ?2, ?3, ?4)", params![name, pass_hash, admin as i64, now()], )?; Ok(conn.last_insert_rowid()) } pub fn user_by_name(&self, name: &str) -> Result> { self.one_user("SELECT id, name, pass_hash, is_admin FROM users WHERE name = ?1", name) } pub fn user_by_id(&self, id: i64) -> Result> { self.one_user("SELECT id, name, pass_hash, is_admin FROM users WHERE id = ?1", id) } fn one_user(&self, sql: &str, key: P) -> Result> { let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare(sql)?; let mut rows = stmt.query(params![key])?; Ok(match rows.next()? { Some(r) => Some(User { id: r.get(0)?, name: r.get(1)?, pass_hash: r.get(2)?, is_admin: r.get::<_, i64>(3)? != 0, }), None => None, }) } pub fn users(&self) -> Result> { let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare("SELECT id, name, pass_hash, is_admin FROM users ORDER BY name")?; let out = stmt .query_map([], |r| { Ok(User { id: r.get(0)?, name: r.get(1)?, pass_hash: r.get(2)?, is_admin: r.get::<_, i64>(3)? != 0, }) })? .collect::>>()?; Ok(out) } pub fn set_password(&self, id: i64, hash: &str) -> Result<()> { let conn = self.conn.lock().unwrap(); conn.execute("UPDATE users SET pass_hash = ?2 WHERE id = ?1", params![id, hash])?; Ok(()) } pub fn set_admin(&self, id: i64, admin: bool) -> Result<()> { let conn = self.conn.lock().unwrap(); conn.execute("UPDATE users SET is_admin = ?2 WHERE id = ?1", params![id, admin as i64])?; Ok(()) } /// Sessions go with the user: a deleted account must not leave a usable cookie behind. pub fn delete_user(&self, id: i64) -> Result<()> { let conn = self.conn.lock().unwrap(); conn.execute("DELETE FROM sessions WHERE user_id = ?1", [id])?; conn.execute("DELETE FROM users WHERE id = ?1", [id])?; Ok(()) } pub fn create_session(&self, user_id: i64, token: &str) -> Result<()> { let conn = self.conn.lock().unwrap(); conn.execute( "INSERT INTO sessions (token, user_id, created, seen) VALUES (?1, ?2, ?3, ?3)", params![token, user_id, now()], )?; Ok(()) } /// 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 fn session_user(&self, token: &str, max_idle_secs: i64) -> Result> { let conn = self.conn.lock().unwrap(); let cutoff = now() - max_idle_secs; let mut stmt = conn.prepare( "SELECT u.id, u.name, u.pass_hash, u.is_admin FROM sessions s JOIN users u ON u.id = s.user_id WHERE s.token = ?1 AND s.seen >= ?2", )?; let mut rows = stmt.query(params![token, cutoff])?; let found = match rows.next()? { Some(r) => Some(User { id: r.get(0)?, name: r.get(1)?, pass_hash: r.get(2)?, is_admin: r.get::<_, i64>(3)? != 0, }), None => None, }; drop(rows); drop(stmt); if found.is_some() { conn.execute("UPDATE sessions SET seen = ?2 WHERE token = ?1", params![token, now()])?; } else { // Either unknown or timed out; either way it is dead weight. conn.execute("DELETE FROM sessions WHERE token = ?1 OR seen < ?2", params![token, cutoff])?; } Ok(found) } pub fn delete_session(&self, token: &str) -> Result<()> { let conn = self.conn.lock().unwrap(); conn.execute("DELETE FROM sessions WHERE token = ?1", [token])?; 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 fn mark_all_read(&self, user_id: i64, feed_ids: &[String]) -> Result { let conn = self.conn.lock().unwrap(); let mut n = 0; for id in feed_ids { n += conn.execute( "INSERT INTO entry_state (user_id, feed_id, guid, read) SELECT ?1, e.feed_id, e.guid, 1 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 coalesce(s.read, 0) = 0 ON CONFLICT(user_id, feed_id, guid) DO UPDATE SET read = 1", rusqlite::params![user_id, id], )?; } Ok(n) } /// The next N enclosures with no file, newest entry first -- what "download latest" /// queues up. pub fn undownloaded(&self, feed_id: &str, limit: i64) -> Result> { let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare( "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", )?; Ok(stmt .query_map(rusqlite::params![feed_id, limit], |r| r.get(0))? .collect::>>()?) } pub fn enclosure(&self, id: i64) -> Result> { let conn = self.conn.lock().unwrap(); Ok(conn .query_row( "SELECT id, feed_id, guid, url, mime, length, path, state, last_error FROM enclosures WHERE id = ?1", [id], |r| { Ok(EncRow { id: r.get(0)?, feed_id: r.get(1)?, guid: r.get(2)?, url: r.get(3)?, mime: r.get(4)?, length: r.get(5)?, path: r.get(6)?, state: r.get(7)?, last_error: r.get(8)?, }) }, ) .optional()?) } /// Read and starred, per person. The row is created on first touch. pub fn set_entry_flag( &self, user_id: i64, feed_id: &str, guid: &str, field: EntryFlag, on: bool, ) -> Result<()> { let conn = self.conn.lock().unwrap(); let col = match field { EntryFlag::Read => "read", EntryFlag::Flagged => "flagged", }; conn.execute( &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}" ), rusqlite::params![user_id, feed_id, guid, on as i64], )?; 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 fn downloaded_count(&self, feed_id: &str) -> Result { let conn = self.conn.lock().unwrap(); Ok(conn.query_row( "SELECT count(*) FROM enclosures WHERE feed_id = ?1 AND path IS NOT NULL", [feed_id], |r| r.get(0), )?) } /// 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], )?; 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( "INSERT INTO feeds (id, url, title, group_id, managed, orphaned) VALUES (?1, ?2, ?3, ?4, 1, 0) 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], )?; 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( "SELECT id, url, title, group_id, orphaned 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)?, orphaned: r.get::<_, i64>(4)? != 0, }) })? .collect::<rusqlite::Result<Vec<_>>>()?) } /// 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])?; Ok(()) } /// 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])?; Ok(()) } 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], )?; 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 fn requeue_interrupted(&self) -> Result<usize> { let conn = self.conn.lock().unwrap(); Ok(conn.execute( "UPDATE enclosures SET state = 'pending' WHERE state = 'downloading' AND path IS NULL", [], )?) } /// 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 fn requeue(&self, id: i64) -> Result<()> { let conn = self.conn.lock().unwrap(); conn.execute( "UPDATE enclosures SET state = 'pending', last_error = NULL WHERE id = ?1 AND path IS NULL", [id], )?; 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::*; #[test] fn deleting_a_shared_file_asks_about_everyone_else() { let db = Db::memory().unwrap(); db.exec_for_test( "INSERT INTO users (id, name, is_admin, created) VALUES (1,'ray',1,0),(2,'sam',0,0),(3,'kit',0,0); INSERT INTO subscriptions (user_id, feed_id, created) VALUES (1,'f',0),(2,'f',0),(3,'f',0); 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');", ) .unwrap(); // Nobody has touched it: both others still have it unplayed. assert_eq!(db.others_wanting(1, 1).unwrap(), (0, 2)); // Sam reads it, Kit stars it. db.set_entry_flag(2, "f", "a", EntryFlag::Read, true).unwrap(); db.set_entry_flag(3, "f", "a", EntryFlag::Flagged, true).unwrap(); assert_eq!(db.others_wanting(1, 1).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).unwrap(); assert_eq!(db.others_wanting(1, 3).unwrap(), (0, 0)); } #[test] fn read_state_belongs_to_one_person() { let db = Db::memory().unwrap(); db.exec_for_test( "INSERT INTO users (id, name, is_admin, created) VALUES (1,'ray',1,0),(2,'sam',0,0); INSERT INTO entries (feed_id, guid, title, first_seen) VALUES ('f','a','One',100),('f','b','Two',200);", ) .unwrap(); assert_eq!(db.unread_count(1, "f").unwrap(), 2); assert_eq!(db.unread_count(2, "f").unwrap(), 2); db.set_entry_flag(1, "f", "a", EntryFlag::Read, true).unwrap(); assert_eq!(db.unread_count(1, "f").unwrap(), 1, "ray read one of them"); assert_eq!(db.unread_count(2, "f").unwrap(), 2, "sam has read nothing"); // Starring and position are just as private. db.set_entry_flag(1, "f", "b", EntryFlag::Flagged, true).unwrap(); db.set_position(2, "f", "b", 42).unwrap(); let ray = db.entries(1, "f", Filter::All, None, 0, 50).unwrap(); let sam = db.entries(2, "f", Filter::All, None, 0, 50).unwrap(); 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); // Marking a whole feed read is likewise one person's business. assert_eq!(db.mark_all_read(2, &["f".to_string()]).unwrap(), 2); assert_eq!(db.unread_count(2, "f").unwrap(), 0); assert_eq!(db.unread_count(1, "f").unwrap(), 1); } #[test] fn schema_is_idempotent_and_summary_handles_unknown_feeds() { let db = Db::memory().unwrap(); // Re-running the schema must not fail: open() does this on every start. db.conn.lock().unwrap().execute_batch(SCHEMA).unwrap(); let sum = db.feed_summary("never-seen").unwrap(); assert_eq!(sum.last_checked, None); assert_eq!(sum.entries, 0); assert_eq!(sum.downloaded, 0); } #[test] 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().unwrap(); db.exec_for_test( "INSERT INTO entries (feed_id, guid, title, description, first_seen) VALUES ('f','a','Alpha dive','notes one',100), ('f','b','Beta', 'notes two',200), ('f','c','Gamma dive','notes three',300); 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, created) VALUES (7,'reader',1,0); INSERT INTO entry_state (user_id, feed_id, guid, read, flagged) VALUES (7,'f','b',1,0), (7,'f','c',1,1);", ) .unwrap(); for f in [Filter::All, Filter::Unread, Filter::Downloaded, Filter::Flagged] { // Both paths must run without erroring, and agree with each other. let rows = db.entries(7, "f", f, None, 0, 50).unwrap(); let n = db.count_entries(7, "f", f, None).unwrap(); assert_eq!(rows.len() as i64, n, "{f:?} count disagrees with the page"); let rows = db.entries(7, "f", f, Some("dive"), 0, 50).unwrap(); let n = db.count_entries(7, "f", f, Some("dive")).unwrap(); assert_eq!(rows.len() as i64, n, "{f:?} with search disagrees"); } assert_eq!(db.count_entries(7, "f", Filter::All, None).unwrap(), 3); assert_eq!(db.count_entries(7, "f", Filter::Unread, None).unwrap(), 1); assert_eq!(db.count_entries(7, "f", Filter::Downloaded, None).unwrap(), 1); assert_eq!(db.count_entries(7, "f", Filter::Flagged, None).unwrap(), 1); assert_eq!(db.count_entries(7, "f", Filter::All, Some("dive")).unwrap(), 2); assert_eq!(db.count_entries(7, "f", Filter::All, Some("NOTES two")).unwrap(), 1, "search is case-insensitive and covers the description"); } #[test] 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().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');", ) .unwrap(); let got: Vec<String> = db.pending("f", 2).unwrap().into_iter().map(|p| p.url).collect(); assert_eq!(got, vec!["u-new", "u-mid"], "newest first, oldest left for later"); } #[test] fn a_restart_requeues_interrupted_downloads() { let db = Db::memory().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');", ) .unwrap(); assert_eq!(db.requeue_interrupted().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"); } #[test] fn enclosure_url_is_the_dedupe_key() { let db = Db::memory().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"); } }