rays/ipodderx-rs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
5362436766696aed4db0fe1ef8b14e2c4b73d7c2
ipodderx-rs/src/db.rs
rays 5362436766 Sortable item table, Size in its own column, Subscribed as an icon 
- Every column heading sorts (kept, title, feed, file type, size,
  published); a second click reverses it. The server sorts through a
  fixed whitelist (order_sql), so it covers the whole list, not the
  fifty loaded; the choice is remembered in the browser.
- Size is its own column and shows KB for small files instead of
  "0 MB". The Item heading is Title.
- Popular/Directory/Add feed: Subscribed is a green circle-check.
- Tests: every sort column runs and orders both ways (db); the table
  sorts by title both ways and remembers across a reload (browser).

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0173mGu6rK18Ne7UGTwAaVJV
2026-09-11 17:17:16 +00:00

1573 lines
62 KiB
Rust
Raw Blame History

//! 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<Connection>,
}
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<Vec<String>>,
pub auto_download: Option<bool>,
pub allow_explicit: Option<bool>,
pub max_new_per_check: Option<i64>,
}
/// 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<String>,
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<String>,
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<String> = stmt
.query_map([], |r| r.get::<_, String>(1))?
.collect::<rusqlite::Result<Vec<_>>>()?;
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<String>,
pub image: Option<String>,
/// 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<i64>,
pub last_error: Option<String>,
pub entries: i64,
pub downloaded: i64,
}
impl Db {
pub fn open(path: &Path) -> Result<Self> {
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<Self> {
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<FeedSummary> {
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<String>,
pub last_modified: Option<String>,
pub last_checked: Option<i64>,
pub ttl_mins: Option<u64>,
}
impl Db {
pub fn http_state(&self, feed_id: &str) -> Result<HttpState> {
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<i64>>(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<u64>,
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<bool> {
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<bool> {
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<String>,
}
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<Vec<Pending>> {
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::<rusqlite::Result<Vec<_>>>()?;
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<Vec<Candidate>> {
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::<rusqlite::Result<Vec<_>>>()?;
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<Vec<(i64, String)>> {
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::<rusqlite::Result<Vec<_>>>()?;
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<usize> {
let conn = self.conn.lock().unwrap();
let n = conn.execute(
"DELETE FROM entries
WHERE coalesce(published, first_seen) < ?1
AND NOT EXISTS (
SELECT 1 FROM enclosures e
WHERE e.feed_id = entries.feed_id AND e.guid = entries.guid
AND e.path IS NOT NULL)
-- Starred by anyone keeps it, the same rule the reaper follows.
AND NOT EXISTS (
SELECT 1 FROM entry_state s
WHERE s.feed_id = entries.feed_id AND s.guid = entries.guid
AND s.flagged = 1)",
[older_than],
)?;
// Whatever went takes everyone's read state with it, rather than leaving rows
// pointing at an item that no longer exists.
conn.execute(
"DELETE FROM entry_state WHERE NOT EXISTS (
SELECT 1 FROM entries e
WHERE e.feed_id = entry_state.feed_id AND e.guid = entry_state.guid)",
[],
)?;
Ok(n)
}
}
impl Db {
pub fn unread_count(&self, user_id: i64, feed_id: &str) -> Result<i64> {
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<String>,
pub link: Option<String>,
pub published: Option<i64>,
pub description: Option<String>,
pub read: bool,
pub flagged: bool,
pub image: Option<String>,
pub duration: Option<i64>,
pub episode: Option<i64>,
pub season: Option<i64>,
pub position: i64,
pub enclosures: Vec<EncRow>,
}
/// 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 feeds a query covers: one, or every feed the person subscribes to. Both forms
/// mention `?1`, since binding a parameter the statement does not use is an error.
fn scope_sql(feed_id: Option<&str>, user_param: u8) -> String {
match feed_id {
Some(_) => "e.feed_id = ?1".into(),
None => format!(
"?1 IS NULL AND e.feed_id IN (SELECT feed_id FROM subscriptions WHERE user_id = ?{user_param})"
),
}
}
/// 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.
pub fn order_sql(col: &str, dir: &str) -> String {
let expr = match col {
"kept" => "coalesce(s.flagged, 0)",
"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)",
};
let dir = if dir == "asc" { "ASC" } else { "DESC" };
format!("{expr} {dir}, coalesce(e.published, e.first_seen) DESC, e.rowid DESC")
}
/// 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<String>,
pub length: Option<i64>,
pub path: Option<String>,
pub state: String,
pub last_error: Option<String>,
}
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<Vec<EntryRow>> {
self.entries_in(user_id, Some(feed_id), filter, search, offset, limit, &order_sql("published", "desc"))
}
/// `entries` for one feed, or across every feed the person subscribes to when `feed_id`
/// is None: the All Subscriptions view.
pub fn entries_in(
&self,
user_id: i64,
feed_id: Option<&str>,
filter: Filter,
search: Option<&str>,
offset: i64,
limit: i64,
order: &str,
) -> Result<Vec<EntryRow>> {
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 {} AND {} AND {SEARCH}
ORDER BY {order}
LIMIT ?4 OFFSET ?3",
scope_sql(feed_id, 5),
filter.sql()
);
let mut stmt = conn.prepare(&sql)?;
let map = |r: &rusqlite::Row| -> rusqlite::Result<EntryRow> {
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<EntryRow> = stmt
.query_map(rusqlite::params![feed_id, like, offset, limit, user_id], map)?
.collect::<rusqlite::Result<Vec<_>>>()?;
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 placeholders = std::iter::repeat_n("?", rows.len()).collect::<Vec<_>>().join(",");
let sql = format!(
"SELECT id, feed_id, guid, url, mime, length, path, state, last_error
FROM enclosures WHERE guid IN ({placeholders}) ORDER BY id"
);
let mut params: Vec<&dyn rusqlite::ToSql> = Vec::with_capacity(rows.len());
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::<rusqlite::Result<Vec<_>>>()?;
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(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<i64> {
self.count_in(user_id, Some(feed_id), filter, search)
}
/// `count_entries` for one feed, or across every feed the person subscribes to.
pub fn count_in(
&self,
user_id: i64,
feed_id: Option<&str>,
filter: Filter,
search: Option<&str>,
) -> Result<i64> {
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 {} AND {} AND {SEARCH}",
scope_sql(feed_id, 3),
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<usize> {
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<Option<Sub>> {
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<String>>(0)?
.and_then(|j| serde_json::from_str(&j).ok()),
auto_download: r.get::<_, Option<i64>>(1)?.map(|v| v != 0),
allow_explicit: r.get::<_, Option<i64>>(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<Vec<Sub>> {
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<String>>(1)?
.and_then(|j| serde_json::from_str(&j).ok()),
auto_download: r.get::<_, Option<i64>>(2)?.map(|v| v != 0),
allow_explicit: r.get::<_, Option<i64>>(3)?.map(|v| v != 0),
max_new_per_check: r.get(4)?,
})
})?
.collect::<rusqlite::Result<Vec<_>>>()?;
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<Vec<Sub>> {
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<String>>(0)?
.and_then(|j| serde_json::from_str(&j).ok()),
auto_download: r.get::<_, Option<i64>>(1)?.map(|v| v != 0),
allow_explicit: r.get::<_, Option<i64>>(2)?.map(|v| v != 0),
max_new_per_check: r.get(3)?,
})
})?
.collect::<rusqlite::Result<Vec<_>>>()?;
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<std::collections::HashMap<String, i64>> {
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::<rusqlite::Result<std::collections::HashMap<_, _>>>()?;
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<i64>>(0)?.unwrap_or(0), r.get::<_, Option<i64>>(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<i64> {
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<Vec<String>> {
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::<rusqlite::Result<Vec<_>>>()?;
Ok(out)
}
// ---- users and sessions ----
pub fn create_user(&self, name: &str, pass_hash: Option<&str>, admin: bool) -> Result<i64> {
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<Option<User>> {
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<Option<User>> {
self.one_user("SELECT id, name, pass_hash, is_admin FROM users WHERE id = ?1", id)
}
fn one_user<P: rusqlite::ToSql>(&self, sql: &str, key: P) -> Result<Option<User>> {
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<Vec<User>> {
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::<rusqlite::Result<Vec<_>>>()?;
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<Option<User>> {
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<usize> {
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<Vec<i64>> {
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::<rusqlite::Result<Vec<_>>>()?)
}
pub fn enclosure(&self, id: i64) -> Result<Option<EncRow>> {
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<i64> {
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 <head><title>.
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 every_sort_column_runs_and_orders_both_ways() {
let db = Db::memory().unwrap();
db.exec_for_test(
"INSERT INTO users (id, name, is_admin, created) VALUES (1,'ray',1,0);
INSERT INTO subscriptions (user_id, feed_id, created) VALUES (1,'f',0),(1,'g',0);
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');",
)
.unwrap();
let order = |col: &str, dir: &str| -> Vec<String> {
db.entries_in(1, None, Filter::All, None, 0, 50, &order_sql(col, dir))
.unwrap()
.into_iter()
.map(|e| e.guid)
.collect()
};
assert_eq!(order("title", "asc"), ["b", "a", "c"], "Apple, banana, cherry: case folded");
assert_eq!(order("title", "desc"), ["c", "a", "b"]);
assert_eq!(order("feed", "asc"), ["b", "c", "a"], "Aardvark, then Zebra's newest first");
assert_eq!(order("type", "asc"), ["a", "b", "c"], "audio, image, video");
assert_eq!(order("size", "desc"), ["c", "a", "b"]);
assert_eq!(order("published", "desc"), ["c", "b", "a"]);
db.set_entry_flag(1, "f", "a", EntryFlag::Flagged, true).unwrap();
assert_eq!(order("kept", "desc")[0], "a");
// An unknown column or direction is newest first; the name itself never reaches the SQL.
assert_eq!(order("title; DROP TABLE entries", "sideways"), ["c", "b", "a"]);
assert!(!order_sql("x'; --", "asc").contains("x'"));
}
#[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");
}
}
Reference in New Issue View Git Blame Copy Permalink