//! SQLite state. Replaces the per-feed .ipxd plists, history.dat and qmcache.dat. use anyhow::{Context, Result}; use rusqlite::{Connection, OptionalExtension}; 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, etag TEXT, last_modified TEXT, last_checked INTEGER, ttl_mins INTEGER, last_error TEXT ); 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, 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); "; /// What `ipx list` shows next to each configured feed. #[derive(Debug, Default)] pub struct FeedSummary { pub title: Option, 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")?; 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)?; 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, last_checked, last_error FROM feeds WHERE id = ?1", [feed_id], |r| { Ok(FeedSummary { title: r.get(0)?, last_checked: r.get(1)?, last_error: r.get(2)?, ..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, ) -> Result<()> { let conn = self.conn.lock().unwrap(); conn.execute( "INSERT INTO feeds (id, url, title, etag, last_modified, last_checked, ttl_mins, last_error) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, NULL) 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, last_error = NULL", rusqlite::params![feed_id, url, title, etag, last_modified, now(), ttl_mins.map(|t| t as i64)], )?; 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(()) } 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) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, 0, 0)", rusqlite::params![feed_id, e.guid, e.title, e.link, e.published, e.description, now()], )?; 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), 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], )?; } 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 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( "SELECT url, mime FROM enclosures WHERE feed_id = ?1 AND state = 'pending' ORDER BY id LIMIT ?2", )?; let rows = stmt .query_map(rusqlite::params![feed_id, limit as i64], |r| { Ok(Pending { url: r.get(0)?, mime: r.get(1)? }) })? .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, flagged ones excluded, read before unread and oldest first within /// each group. /// /// 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). Requiring /// `read = 1` outright would be just as dead here, since nothing marks episodes read /// until a UI exists -- so `flagged` is the keep-forever marker, and `read` only decides /// what goes first. 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), coalesce(n.read, 0), coalesce(n.flagged, 0) FROM enclosures e LEFT JOIN entries n ON n.feed_id = e.feed_id AND n.guid = e.guid WHERE e.path IS NOT NULL AND coalesce(n.flagged, 0) = 0 ORDER BY coalesce(n.read, 0) 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) } } /// 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 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 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"); } }