rays/ipodderx-rs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
9aae3097e771176fd2107d4c3bebf5aac06f0a0f
ipodderx-rs/src/db.rs
rays 586d2c07a1 ipx user rename: give an account the name the proxy signs it in as 
An account made by hand before the proxy was set up is called what it was
given ('rays'), while Cloudflare Access vouches for an email address. With
auto_create_users on, the first visit through the tunnel would make a
second, empty account. Renaming keeps the id, so feeds, read state and
admin rights go with it.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TAC7sLVqfKmY6rsTLXzNgk
2026-09-12 13:54:29 +00:00

1778 lines
72 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,
image TEXT,
duration INTEGER,
episode INTEGER,
season INTEGER,
PRIMARY KEY (feed_id, guid)
);
-- url is UNIQUE: this is the dedupe key, and it subsumes the old history.dat pickle.
-- A reaped file keeps its row with path = NULL and state = 'reaped', so a purged
-- episode is never fetched a second time.
CREATE TABLE IF NOT EXISTS enclosures (
id INTEGER PRIMARY KEY,
feed_id TEXT NOT NULL,
guid TEXT NOT NULL,
url TEXT NOT NULL UNIQUE,
mime TEXT,
length INTEGER,
path TEXT,
state TEXT NOT NULL,
bytes_done INTEGER NOT NULL DEFAULT 0,
downloaded_at INTEGER,
last_error TEXT
);
CREATE INDEX IF NOT EXISTS enclosures_entry ON enclosures (feed_id, guid);
-- pass_hash is NULL for someone who only ever arrives through the proxy: there is no
-- password to check, and leaving it empty is not the same as leaving it unset.
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL UNIQUE COLLATE NOCASE,
pass_hash TEXT,
is_admin INTEGER NOT NULL DEFAULT 0,
-- For whoever maintains the server. NULL where it is not known.
created INTEGER,
last_login INTEGER
);
-- 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,
PRIMARY KEY (user_id, feed_id)
);
-- Read, kept and how far in. One row per person per item, created on first touch;
-- an item nobody has touched has no row at all, which is what unread means.
CREATE TABLE IF NOT EXISTS entry_state (
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
feed_id TEXT NOT NULL,
guid TEXT NOT NULL,
read INTEGER NOT NULL DEFAULT 0,
flagged INTEGER NOT NULL DEFAULT 0,
position INTEGER NOT NULL DEFAULT 0,
PRIMARY KEY (user_id, feed_id, guid)
);
CREATE TABLE IF NOT EXISTS sessions (
token TEXT PRIMARY KEY,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
seen INTEGER NOT NULL
);
";
/// One person's wants for one feed. `None` in a field means the feed's own setting stands.
#[derive(Debug, Clone, Default)]
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,
/// When the account was made and when it last signed in, for whoever maintains the server.
pub created: Option<i64>,
pub last_login: Option<i64>,
}
/// The columns `user_row` reads, in its order.
const USER_COLS: &str = "id, name, pass_hash, is_admin, created, last_login";
fn user_row(r: &rusqlite::Row<'_>) -> rusqlite::Result<User> {
Ok(User {
id: r.get(0)?,
name: r.get(1)?,
pass_hash: r.get(2)?,
is_admin: r.get::<_, i64>(3)? != 0,
created: r.get(4)?,
last_login: r.get(5)?,
})
}
/// 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,
}
/// Adds the columns later versions introduced and drops the ones they retired. CREATE TABLE IF
/// NOT EXISTS leaves a table that already exists alone, so an installed database needs both done
/// explicitly. Columns from before 0.3.0, the oldest version an upgrade may start from, need no
/// entry.
fn migrate(conn: &Connection) -> Result<()> {
let wanted: &[(&str, &str, &str)] = &[
// For whoever maintains the server. An audit dropped `created` as unread on 2026-09-12,
// and it came back the same day with `last_login` beside it.
("users", "created", "INTEGER"),
("users", "last_login", "INTEGER"),
];
let retired: &[(&str, &str)] = &[
// Read state from before accounts, long since moved to entry_state. Two bugs came from
// queries still reading these after they stopped meaning anything.
("entries", "read"),
("entries", "flagged"),
("entries", "position"),
// Written by every insert and read by nothing.
("subscriptions", "created"),
("sessions", "created"),
];
let has = |table: &str, column: &str| -> Result<bool> {
let names: Vec<String> = conn
.prepare(&format!("PRAGMA table_info({table})"))?
.query_map([], |r| r.get(1))?
.collect::<rusqlite::Result<_>>()?;
Ok(names.iter().any(|c| c == column))
};
for (table, column, ty) in wanted {
if !has(table, column)? {
tracing::info!(table, column, "adding column");
conn.execute_batch(&format!("ALTER TABLE {table} ADD COLUMN {column} {ty}"))?;
}
}
for (table, column) in retired {
if has(table, column)? {
tracing::info!(table, column, "dropping column");
conn.execute_batch(&format!("ALTER TABLE {table} DROP COLUMN {column}"))?;
}
}
Ok(())
}
/// What `ipx list` shows next to each configured feed.
#[derive(Debug, Default)]
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.
///
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,
image, duration, episode, season)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)",
rusqlite::params![
feed_id, e.guid, e.title, e.link, e.published, e.description, now(),
e.image, e.duration, e.episode, e.season
],
)?;
if inserted == 0 {
conn.execute(
"UPDATE entries SET
title = coalesce(?3, title),
description = coalesce(?4, description),
image = coalesce(?5, image),
duration = coalesce(?6, duration),
episode = coalesce(?7, episode),
season = coalesce(?8, season)
WHERE feed_id = ?1 AND guid = ?2",
rusqlite::params![
feed_id, e.guid, e.title, e.description,
e.image, e.duration, e.episode, e.season
],
)?;
}
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 entries, each with its enclosures attached: one feed's, or every feed the
/// person subscribes to when `feed_id` is None (All Subscriptions). `search` matches title
/// and description, case-insensitively.
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 `entries_in` would page through, so the UI knows whether there is more.
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(())
}
/// The first admin starts subscribed to the whole catalogue: whoever wrote config.toml meant
/// to read those feeds, and without this a fresh install signs in to an empty sidebar. Runs
/// only while nobody subscribes to anything, so an unsubscribe is never undone.
pub fn adopt_catalogue(&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);
}
for id in catalogue {
conn.execute(
"INSERT OR IGNORE INTO subscriptions (user_id, feed_id) VALUES (?1, ?2)",
params![user_id, id],
)?;
}
// Feeds that exist only in the database (OPML children) count too.
conn.execute(
"INSERT OR IGNORE INTO subscriptions (user_id, feed_id) SELECT ?1, id FROM feeds",
params![user_id],
)?;
Ok(catalogue.len())
}
// ---- 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.
/// In a group, whatever someone has not set on the feed itself comes from their
/// subscription to the group, as the group's settings dialog has always said it does.
pub fn subscribers(&self, feed_id: &str, group: Option<&str>) -> Result<Vec<Sub>> {
let conn = self.conn.lock().unwrap();
let mut stmt = conn.prepare(
"SELECT coalesce(c.keywords, p.keywords), coalesce(c.auto_download, p.auto_download),
coalesce(c.allow_explicit, p.allow_explicit),
coalesce(c.max_new_per_check, p.max_new_per_check)
FROM subscriptions c
LEFT JOIN subscriptions p ON p.user_id = c.user_id AND p.feed_id = ?2
WHERE c.feed_id = ?1",
)?;
let out = stmt
.query_map(params![feed_id, group], |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) VALUES (?1, ?2)",
params![user_id, feed_id],
)?;
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(())
}
/// 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)
VALUES (?1, ?2, ?3, ?4, ?5, ?6)
ON CONFLICT(user_id, feed_id) DO UPDATE SET
keywords = excluded.keywords,
auto_download = excluded.auto_download,
allow_explicit = excluded.allow_explicit,
max_new_per_check = excluded.max_new_per_check",
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,
],
)?;
Ok(())
}
// ---- 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(&format!("SELECT {USER_COLS} FROM users WHERE name = ?1"), name)
}
pub fn user_by_id(&self, id: i64) -> Result<Option<User>> {
self.one_user(&format!("SELECT {USER_COLS} 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_row(r)?),
None => None,
})
}
pub fn users(&self) -> Result<Vec<User>> {
let conn = self.conn.lock().unwrap();
let mut stmt =
conn.prepare(&format!("SELECT {USER_COLS} FROM users ORDER BY name"))?;
let out = stmt
.query_map([], user_row)?
.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(())
}
/// The proxy signs people in by the name it vouches for, so an account made before the proxy
/// was set up has to take that name to be found by it. The name is UNIQUE, so a taken one is
/// refused here as well as by the caller.
pub fn rename_user(&self, id: i64, name: &str) -> Result<()> {
let conn = self.conn.lock().unwrap();
conn.execute("UPDATE users SET name = ?2 WHERE id = ?1", params![id, name])?;
Ok(())
}
/// Records a sign-in, to the hour: the proxy vouches for every request, and writing each one
/// would buy nothing.
pub fn signed_in(&self, id: i64) -> Result<()> {
let conn = self.conn.lock().unwrap();
conn.execute(
"UPDATE users SET last_login = ?2 WHERE id = ?1 AND coalesce(last_login, 0) <= ?2 - 3600",
params![id, now()],
)?;
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, seen) VALUES (?1, ?2, ?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, u.created, u.last_login
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_row(r)?),
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 kept, 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 FROM feeds
WHERE managed = 1 AND group_id IS NOT NULL ORDER BY coalesce(title, id)",
)?;
Ok(stmt
.query_map([], |r| {
Ok(Managed {
id: r.get(0)?,
url: r.get(1)?,
title: r.get(2)?,
group_id: r.get(3)?,
})
})?
.collect::<rusqlite::Result<Vec<_>>>()?)
}
/// 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(())
}
/// Empties a feed of its items, leaving its files alone.
pub fn clear_entries(&self, feed_id: &str) -> Result<()> {
let conn = self.conn.lock().unwrap();
conn.execute("DELETE FROM entries WHERE feed_id = ?1", [feed_id])?;
Ok(())
}
/// Every feed the database holds rows for, as (id, url), removed ones included.
pub fn feed_urls(&self) -> Result<Vec<(String, String)>> {
let conn = self.conn.lock().unwrap();
let mut stmt = conn.prepare("SELECT id, coalesce(url, '') FROM feeds")?;
let out = stmt
.query_map([], |r| Ok((r.get(0)?, r.get(1)?)))?
.collect::<rusqlite::Result<_>>()?;
Ok(out)
}
/// Hands a feed in a group the enclosures its parent holds, as (guid, url), with everyone's
/// read state for them. A Patreon creator read as one feed before it was split into shows
/// owns every show's files, and `enclosures.url` is unique, so without this each show would
/// list its items with nothing to play.
pub fn adopt(&self, parent: &str, child: &str, listed: &[(&str, &str)]) -> Result<()> {
let mut conn = self.conn.lock().unwrap();
let holds: bool = conn.query_row(
"SELECT EXISTS (SELECT 1 FROM enclosures WHERE feed_id = ?1)",
[parent],
|r| r.get(0),
)?;
if !holds {
return Ok(()); // An OPML, or a creator already shared out.
}
let tx = conn.transaction()?;
for &(guid, url) in listed {
let moved = tx.execute(
"UPDATE enclosures SET feed_id = ?3, guid = ?4 WHERE url = ?1 AND feed_id = ?2",
params![url, parent, child, guid],
)?;
if moved == 1 {
// Patreon gives a post the same guid in every feed it appears in.
tx.execute(
"UPDATE OR IGNORE entry_state SET feed_id = ?2 WHERE feed_id = ?1 AND guid = ?3",
params![parent, child, guid],
)?;
}
}
tx.commit()?;
Ok(())
}
/// A feed's enclosures skipped by one of its filters, by URL, with the reason: the verdicts a
/// change of settings can overturn. A torrent held back while torrents are off is not a
/// filter's call.
pub fn skipped_by_filter(&self, feed_id: &str) -> Result<std::collections::HashMap<String, String>> {
let conn = self.conn.lock().unwrap();
let mut stmt = conn.prepare(
"SELECT url, last_error FROM enclosures
WHERE feed_id = ?1 AND state = 'skipped' AND last_error IS NOT NULL
AND last_error != 'torrents disabled'",
)?;
let out = stmt
.query_map([feed_id], |r| Ok((r.get(0)?, r.get(1)?)))?
.collect::<rusqlite::Result<_>>()?;
Ok(out)
}
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) VALUES (1,'ray',1);
INSERT INTO subscriptions (user_id, feed_id) VALUES (1,'f'),(1,'g');
INSERT INTO feeds (id, url, title) VALUES ('f','u','Zebra'),('g','v','Aardvark');
INSERT INTO entries (feed_id, guid, title, first_seen) VALUES
('f','a','banana',100),('g','b','Apple',200),('f','c','cherry',300);
INSERT INTO enclosures (id, feed_id, guid, url, mime, length, state) VALUES
(1,'f','a','u1','audio/mpeg',300,'pending'),(2,'g','b','u2','image/png',10,'pending'),
(3,'f','c','u3','video/mp4',2000,'pending');",
)
.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) VALUES (1,'ray',1),(2,'sam',0),(3,'kit',0);
INSERT INTO subscriptions (user_id, feed_id) VALUES (1,'f'),(2,'f'),(3,'f');
INSERT INTO entries (feed_id, guid, first_seen) VALUES ('f','a',0);
INSERT INTO enclosures (id, feed_id, guid, url, path, state) VALUES
(1,'f','a','u1','/tmp/a','done');",
)
.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) VALUES (1,'ray',1),(2,'sam',0);
INSERT INTO entries (feed_id, guid, title, first_seen) VALUES
('f','a','One',100),('f','b','Two',200);",
)
.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 order = order_sql("published", "desc");
let page = |user| db.entries_in(user, Some("f"), Filter::All, None, 0, 50, &order).unwrap();
let (ray, sam) = (page(1), page(2));
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 an_old_database_loses_its_retired_columns() {
let conn = Connection::open_in_memory().unwrap();
// As open() has it: a DROP COLUMN on a table that references another is the part worth
// proving, and it has to work with the foreign keys switched on.
conn.pragma_update(None, "foreign_keys", "ON").unwrap();
conn.execute_batch(
"CREATE TABLE entries (feed_id TEXT NOT NULL, guid TEXT NOT NULL,
first_seen INTEGER NOT NULL, read INTEGER NOT NULL DEFAULT 0,
flagged INTEGER NOT NULL DEFAULT 0, position INTEGER NOT NULL DEFAULT 0,
PRIMARY KEY (feed_id, guid));
CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT NOT NULL UNIQUE COLLATE NOCASE,
pass_hash TEXT, is_admin INTEGER NOT NULL DEFAULT 0, created INTEGER NOT NULL);
CREATE TABLE subscriptions (
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
feed_id TEXT NOT NULL, created INTEGER NOT NULL, PRIMARY KEY (user_id, feed_id));
CREATE TABLE sessions (token TEXT PRIMARY KEY,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
created INTEGER NOT NULL, seen INTEGER NOT NULL);
INSERT INTO users VALUES (1, 'ray', NULL, 1, 0);
INSERT INTO subscriptions VALUES (1, 'f', 0);
INSERT INTO sessions VALUES ('t', 1, 0, 0);",
)
.unwrap();
// The same order as open(): the schema leaves the old tables alone, migrate() fixes them.
conn.execute_batch(SCHEMA).unwrap();
migrate(&conn).unwrap();
let cols = |table: &str| -> Vec<String> {
conn.prepare(&format!("PRAGMA table_info({table})"))
.unwrap()
.query_map([], |r| r.get(1))
.unwrap()
.collect::<rusqlite::Result<_>>()
.unwrap()
};
for (table, gone) in [
("entries", &["read", "flagged", "position"][..]),
("subscriptions", &["created"][..]),
("sessions", &["created"][..]),
] {
let cols = cols(table);
assert!(!cols.iter().any(|c| gone.contains(&c.as_str())), "{table}: {cols:?}");
}
// users.created is not retired: it keeps what it held, and last_login joins it.
let users = cols("users");
assert!(users.iter().any(|c| c == "last_login"), "{users:?}");
assert_eq!(conn.query_row("SELECT created FROM users", [], |r| r.get::<_, i64>(0)).unwrap(), 0);
// And the rows come through it.
let kept: i64 = conn
.query_row("SELECT count(*) FROM subscriptions JOIN sessions USING (user_id)", [], |r| r.get(0))
.unwrap();
assert_eq!(kept, 1);
}
#[test]
fn a_renamed_account_keeps_everything_but_its_name() {
let db = Db::memory().unwrap();
let ray = db.create_user("rays", None, true).unwrap();
db.create_user("sam", None, false).unwrap();
db.subscribe(ray, "f").unwrap();
db.rename_user(ray, "rays@sdf1.net").unwrap();
assert!(db.user_by_name("rays").unwrap().is_none());
let renamed = db.user_by_name("RAYS@sdf1.net").unwrap().unwrap();
assert_eq!((renamed.id, renamed.is_admin), (ray, true), "same account, still the admin");
assert_eq!(db.subscriptions_for(ray).unwrap().len(), 1, "and still subscribed");
assert!(db.rename_user(ray, "sam").is_err(), "a taken name is refused");
}
#[test]
fn an_account_knows_when_it_was_made_and_last_signed_in() {
let db = Db::memory().unwrap();
let id = db.create_user("ray", None, true).unwrap();
let get = || db.user_by_id(id).unwrap().unwrap();
assert!(get().created.is_some_and(|t| t > 0));
assert_eq!(get().last_login, None, "made, but never signed in");
db.signed_in(id).unwrap();
let first = get().last_login.unwrap();
// Within the hour, the proxy vouching again writes nothing; after it, it does.
db.exec_for_test(&format!("UPDATE users SET last_login = {} WHERE id = {id}", first - 60)).unwrap();
db.signed_in(id).unwrap();
assert_eq!(get().last_login, Some(first - 60));
db.exec_for_test(&format!("UPDATE users SET last_login = {} WHERE id = {id}", first - 7200)).unwrap();
db.signed_in(id).unwrap();
assert!(get().last_login.unwrap() >= first);
}
#[test]
fn the_first_admin_starts_with_the_catalogue_and_only_once() {
// Cutting this along with the dead read columns left the browser suite's admin with an
// empty sidebar: it is how a fresh install's first account gets config.toml's feeds.
let db = Db::memory().unwrap();
db.exec_for_test("INSERT INTO users (id, name, is_admin) VALUES (1,'admin',1);")
.unwrap();
let subs = || -> i64 {
db.conn.lock().unwrap().query_row("SELECT count(*) FROM subscriptions", [], |r| r.get(0)).unwrap()
};
let catalogue = ["a".to_string(), "b".to_string()];
assert_eq!(db.adopt_catalogue(1, &catalogue).unwrap(), 2);
assert_eq!(subs(), 2);
// Once anyone subscribes to anything it never runs again, so an unsubscribe sticks.
db.unsubscribe(1, "a").unwrap();
assert_eq!(db.adopt_catalogue(1, &catalogue).unwrap(), 0);
assert_eq!(subs(), 1);
}
#[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) VALUES (7,'reader',1);
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 order = order_sql("published", "desc");
let rows = db.entries_in(7, Some("f"), f, None, 0, 50, &order).unwrap();
let n = db.count_in(7, Some("f"), f, None).unwrap();
assert_eq!(rows.len() as i64, n, "{f:?} count disagrees with the page");
let rows = db.entries_in(7, Some("f"), f, Some("dive"), 0, 50, &order).unwrap();
let n = db.count_in(7, Some("f"), f, Some("dive")).unwrap();
assert_eq!(rows.len() as i64, n, "{f:?} with search disagrees");
}
assert_eq!(db.count_in(7, Some("f"), Filter::All, None).unwrap(), 3);
assert_eq!(db.count_in(7, Some("f"), Filter::Unread, None).unwrap(), 1);
assert_eq!(db.count_in(7, Some("f"), Filter::Downloaded, None).unwrap(), 1);
assert_eq!(db.count_in(7, Some("f"), Filter::Flagged, None).unwrap(), 1);
assert_eq!(db.count_in(7, Some("f"), Filter::All, Some("dive")).unwrap(), 2);
assert_eq!(db.count_in(7, Some("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 a_show_takes_over_what_its_creator_held() {
let db = Db::memory().unwrap();
db.exec_for_test(
"INSERT INTO users (id, name, is_admin) VALUES (1,'ray',1);
INSERT INTO enclosures (id, feed_id, guid, url, state, path, last_error) VALUES
(1,'creator','a','u1','done','/x/a.mp3',NULL),
(2,'creator','b','u2','skipped',NULL,'explicit'),
(3,'creator','c','u3','skipped',NULL,'explicit'),
(4,'other','d','u4','skipped',NULL,'torrents disabled');
INSERT INTO entry_state (user_id, feed_id, guid, read) VALUES (1,'creator','a',1);",
)
.unwrap();
db.adopt("creator", "show", &[("a", "u1"), ("b", "u2"), ("d", "u4")]).unwrap();
{
let conn = db.conn.lock().unwrap();
let owner = |id: i64| -> String {
conn.query_row("SELECT feed_id FROM enclosures WHERE id = ?1", [id], |r| r.get(0)).unwrap()
};
assert_eq!(owner(1), "show", "a downloaded file moves with its item");
assert_eq!(owner(2), "show");
assert_eq!(owner(3), "creator", "this show does not list it");
assert_eq!(owner(4), "other", "only the parent's are taken");
let read: String = conn
.query_row("SELECT feed_id FROM entry_state WHERE user_id = 1 AND guid = 'a'", [], |r| r.get(0))
.unwrap();
assert_eq!(read, "show", "what you had read stays read");
}
// Only a filter's verdict can be overturned by a change of settings.
let skipped = db.skipped_by_filter("show").unwrap();
assert_eq!(skipped.get("u2").map(String::as_str), Some("explicit"));
assert_eq!(skipped.len(), 1);
assert!(db.skipped_by_filter("other").unwrap().is_empty(), "torrents disabled is not a filter");
}
#[test]
fn a_feed_in_a_group_follows_your_settings_on_the_group() {
let db = Db::memory().unwrap();
db.exec_for_test(
"INSERT INTO users (id, name, is_admin) VALUES (1,'ray',1),(2,'sam',0);
INSERT INTO subscriptions (user_id, feed_id, allow_explicit) VALUES
(1,'group',1),(1,'show',NULL),(2,'group',1),(2,'show',0);",
)
.unwrap();
let explicit = |group| -> Vec<Option<bool>> {
let mut v: Vec<_> =
db.subscribers("show", group).unwrap().into_iter().map(|s| s.allow_explicit).collect();
v.sort();
v
};
assert_eq!(explicit(Some("group")), [Some(false), Some(true)], "ray inherits; sam's own choice on the show wins");
assert_eq!(explicit(None), [None, Some(false)], "outside a group nothing is inherited");
}
#[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