rays
611716d8b7
The last nineteen functions move to SeaORM: recording feeds, items and enclosures, managed OPML feeds, folding WordPress's repeated files, and handing a Patreon creator's files to its shows. Two SQLite-only forms go: GLOB becomes a LIKE with the underscore escaped (broader, harmlessly: the fold still keys on `_=` and digits), and UPDATE OR IGNORE becomes an UPDATE ... WHERE NOT EXISTS. The two transactions are SeaORM transactions. With nothing left on it, rusqlite goes, with the SQL schema and migrate(). The entities are the schema: create_missing makes whatever tables and indexes a database lacks, from them, with CREATE ... IF NOT EXISTS. Production's schema already has every column migrate() added and none it dropped. Not SeaORM's schema sync, used until now: despite its docs it drops a unique index the entities do not describe, so it dropped users_name_lower on every open. Every `ipx` command then took a write lock, and against a daemon busy writing, `ipx status` -- the healthcheck -- failed 7 times in 15 where the old code failed none. Now 15 in 15, as before. On Postgres it would not have started. WAL is set only when a file is not already in it: setting it takes a lock that cannot wait out a busy daemon. Checked on copies of production: a forced scan of all 162 feeds against the real feeds with no database errors; the feed list, filters, sorts, search and the reaper's candidates against the old code on the same data, earlier in the branch. The column comments from the SQL schema move to the entities. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
1970 lines
84 KiB
Rust
1970 lines
84 KiB
Rust
//! The database, through SeaORM: SQLite today, Postgres to come (issue #18). Replaces the
|
|
//! per-feed .ipxd plists, history.dat and qmcache.dat.
|
|
|
|
use anyhow::{Context, Result};
|
|
use crate::entity::{enclosures, entries, feeds, sessions, subscriptions, users};
|
|
use sea_orm::sea_query::{Expr, Func};
|
|
use sea_orm::{
|
|
ActiveModelTrait, ColumnTrait, ConnectionTrait, EntityTrait, PaginatorTrait, QueryFilter, QueryOrder, Set,
|
|
Statement,
|
|
};
|
|
use std::path::Path;
|
|
|
|
/// A pool of connections, where there was one connection behind a global lock.
|
|
pub struct Db {
|
|
orm: sea_orm::DatabaseConnection,
|
|
/// A test's database file, removed when the test is done with it.
|
|
#[cfg(test)]
|
|
tmp: Option<std::path::PathBuf>,
|
|
}
|
|
|
|
#[cfg(test)]
|
|
impl Drop for Db {
|
|
fn drop(&mut self) {
|
|
if let Some(p) = &self.tmp {
|
|
for ext in ["", "-wal", "-shm"] {
|
|
let _ = std::fs::remove_file(format!("{}{ext}", p.display()));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Creates whatever tables and indexes the database is missing, from the entities in
|
|
/// `crate::entity`, on SQLite or Postgres alike. It only ever creates: an existing table is left
|
|
/// as it is. What an entity cannot say -- an index on an expression, or on two columns --
|
|
/// follows as plain SQL both databases accept.
|
|
///
|
|
/// Not SeaORM's schema sync, which is experimental and, despite its docs, drops a unique index
|
|
/// the entities do not describe: it dropped users_name_lower on every open, so every `ipx`
|
|
/// command took a write lock, and the healthcheck's `ipx status` timed out behind a busy daemon.
|
|
/// On Postgres it would have failed outright, dropping that index as a constraint.
|
|
///
|
|
/// ponytail: tables only, no columns. A column added to an existing table needs its own ALTER
|
|
/// here, as the old migrate() did for SQLite, or sea-orm-migration once there are several. A
|
|
/// database from before 0.7 took its last columns from that migrate(), so it upgrades through
|
|
/// a 0.7 release first.
|
|
async fn create_missing(orm: &sea_orm::DatabaseConnection) -> Result<()> {
|
|
use crate::entity::*;
|
|
use sea_orm::{ConnectionTrait, Schema};
|
|
let schema = Schema::new(orm.get_database_backend());
|
|
for mut table in [
|
|
schema.create_table_from_entity(feeds::Entity),
|
|
schema.create_table_from_entity(entries::Entity),
|
|
schema.create_table_from_entity(enclosures::Entity),
|
|
schema.create_table_from_entity(users::Entity),
|
|
schema.create_table_from_entity(subscriptions::Entity),
|
|
schema.create_table_from_entity(entry_state::Entity),
|
|
schema.create_table_from_entity(sessions::Entity),
|
|
] {
|
|
orm.execute(table.if_not_exists()).await.context("creating the schema")?;
|
|
}
|
|
for sql in [
|
|
"CREATE INDEX IF NOT EXISTS enclosures_entry ON enclosures (feed_id, guid)",
|
|
"CREATE UNIQUE INDEX IF NOT EXISTS users_name_lower ON users (lower(name))",
|
|
] {
|
|
orm.execute_unprepared(sql).await.with_context(|| sql.to_owned())?;
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// sqlx's defaults for SQLite are what ipx wants: foreign keys on, and a five-second wait for a
|
|
/// lock, which is what rusqlite was set to.
|
|
async fn connect(path: &Path) -> Result<sea_orm::DatabaseConnection> {
|
|
let mut opts = sea_orm::ConnectOptions::new(format!("sqlite://{}?mode=rwc", path.display()));
|
|
opts.sqlx_logging(false);
|
|
sea_orm::Database::connect(opts)
|
|
.await
|
|
.with_context(|| format!("opening {}", path.display()))
|
|
}
|
|
|
|
|
|
/// 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>,
|
|
}
|
|
|
|
impl From<enclosures::Model> for EncRow {
|
|
fn from(e: enclosures::Model) -> Self {
|
|
EncRow {
|
|
id: e.id,
|
|
feed_id: e.feed_id,
|
|
guid: e.guid,
|
|
url: e.url,
|
|
mime: e.mime,
|
|
length: e.length,
|
|
path: e.path,
|
|
state: e.state,
|
|
last_error: e.last_error,
|
|
}
|
|
}
|
|
}
|
|
|
|
/// A subscription's keywords, stored as a JSON array.
|
|
fn keywords(json: Option<String>) -> Option<Vec<String>> {
|
|
json.and_then(|j| serde_json::from_str(&j).ok())
|
|
}
|
|
|
|
impl From<subscriptions::Model> for Sub {
|
|
fn from(s: subscriptions::Model) -> Self {
|
|
Sub {
|
|
feed_id: s.feed_id,
|
|
keywords: keywords(s.keywords),
|
|
auto_download: s.auto_download,
|
|
allow_explicit: s.allow_explicit,
|
|
max_new_per_check: s.max_new_per_check,
|
|
}
|
|
}
|
|
}
|
|
|
|
impl From<users::Model> for User {
|
|
fn from(u: users::Model) -> Self {
|
|
User {
|
|
id: u.id,
|
|
name: u.name,
|
|
pass_hash: u.pass_hash,
|
|
is_admin: u.is_admin,
|
|
created: u.created,
|
|
last_login: u.last_login,
|
|
}
|
|
}
|
|
}
|
|
|
|
/// A feed derived from an OPML subscription rather than written into the config.
|
|
#[derive(Debug, Clone)]
|
|
pub struct Managed {
|
|
pub id: String,
|
|
pub url: String,
|
|
pub title: Option<String>,
|
|
pub group_id: String,
|
|
}
|
|
|
|
|
|
/// What `ipx list` shows next to each configured feed.
|
|
#[derive(Debug, Default)]
|
|
pub struct FeedSummary {
|
|
pub title: Option<String>,
|
|
pub image: Option<String>,
|
|
pub category: 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>,
|
|
/// When this run of failures began; see the `error_since` column.
|
|
pub error_since: Option<i64>,
|
|
pub entries: i64,
|
|
pub downloaded: i64,
|
|
}
|
|
|
|
impl Db {
|
|
pub async 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 orm = connect(path).await?;
|
|
// WAL, so the healthcheck's `ipx status` reads while the daemon writes. It is a setting of
|
|
// the file, kept once made, and making it takes a lock that cannot wait out a busy
|
|
// daemon, so it is made only when the file is not already in WAL.
|
|
let mode = orm.query_one_raw(Statement::from_string(orm.get_database_backend(), "PRAGMA journal_mode")).await?;
|
|
if mode.and_then(|r| r.try_get_by_index::<String>(0).ok()).as_deref() != Some("wal") {
|
|
orm.execute_unprepared("PRAGMA journal_mode = WAL").await.context("switching to WAL")?;
|
|
}
|
|
create_missing(&orm).await?;
|
|
Ok(Self {
|
|
orm,
|
|
#[cfg(test)]
|
|
tmp: None,
|
|
})
|
|
}
|
|
|
|
/// A fresh database for a test, in a file of its own: two connections to one ":memory:"
|
|
/// would be two databases. Its schema comes from the entities, as a real one's does.
|
|
#[cfg(test)]
|
|
pub async fn memory() -> Result<Self> {
|
|
static N: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0);
|
|
let path = std::env::temp_dir().join(format!(
|
|
"ipx-test-{}-{}.db",
|
|
std::process::id(),
|
|
N.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
|
|
));
|
|
let orm = connect(&path).await?;
|
|
create_missing(&orm).await?;
|
|
Ok(Self { orm, tmp: Some(path) })
|
|
}
|
|
|
|
#[cfg(test)]
|
|
/// The first column of every row, for a test to check what a query left behind.
|
|
#[cfg(test)]
|
|
pub async fn i64s_for_test(&self, sql: &str) -> Vec<i64> {
|
|
self.rows(sql, vec![]).await.unwrap().iter().map(|r| r.try_get_by_index(0).unwrap()).collect()
|
|
}
|
|
|
|
#[cfg(test)]
|
|
pub async fn strings_for_test(&self, sql: &str) -> Vec<String> {
|
|
self.rows(sql, vec![]).await.unwrap().iter().map(|r| r.try_get_by_index(0).unwrap()).collect()
|
|
}
|
|
|
|
#[cfg(test)]
|
|
pub async fn exec_for_test(&self, sql: &str) -> Result<()> {
|
|
self.orm.execute_unprepared(sql).await?;
|
|
Ok(())
|
|
}
|
|
|
|
pub async fn feed_summary(&self, feed_id: &str) -> Result<FeedSummary> {
|
|
let mut sum = feeds::Entity::find_by_id(feed_id.to_owned())
|
|
.one(&self.orm)
|
|
.await?
|
|
.map(|f| FeedSummary {
|
|
title: f.title,
|
|
image: f.image,
|
|
last_checked: f.last_checked,
|
|
last_error: f.last_error,
|
|
orphaned: f.orphaned,
|
|
error_since: f.error_since,
|
|
category: f.category,
|
|
..Default::default()
|
|
})
|
|
.unwrap_or_default();
|
|
sum.entries =
|
|
entries::Entity::find().filter(entries::Column::FeedId.eq(feed_id)).count(&self.orm).await? as i64;
|
|
sum.downloaded = self.downloaded_count(feed_id).await?;
|
|
Ok(sum)
|
|
}
|
|
}
|
|
|
|
|
|
/// Everything `fetch` needs to decide whether to poll a feed, and how.
|
|
#[derive(Debug, Default)]
|
|
pub struct HttpState {
|
|
pub etag: Option<String>,
|
|
pub last_modified: Option<String>,
|
|
pub last_checked: Option<i64>,
|
|
pub ttl_mins: Option<u64>,
|
|
}
|
|
|
|
impl Db {
|
|
pub async fn http_state(&self, feed_id: &str) -> Result<HttpState> {
|
|
Ok(feeds::Entity::find_by_id(feed_id.to_owned())
|
|
.one(&self.orm)
|
|
.await?
|
|
.map(|f| HttpState {
|
|
etag: f.etag,
|
|
last_modified: f.last_modified,
|
|
last_checked: f.last_checked,
|
|
ttl_mins: f.ttl_mins.map(|t| t.max(0) as u64),
|
|
})
|
|
.unwrap_or_default())
|
|
}
|
|
|
|
/// Upsert after a successful poll. Clears any previous error.
|
|
#[allow(clippy::too_many_arguments)]
|
|
pub async fn record_feed(
|
|
&self,
|
|
feed_id: &str,
|
|
url: &str,
|
|
title: Option<&str>,
|
|
etag: Option<&str>,
|
|
last_modified: Option<&str>,
|
|
ttl_mins: Option<u64>,
|
|
image: Option<&str>,
|
|
category: Option<&str>,
|
|
) -> Result<()> {
|
|
// category is taken as it comes, unlike title and image: a show that leaves a category
|
|
// should leave the Directory's chip too.
|
|
let o = |v: Option<&str>| sea_orm::Value::from(v.map(str::to_owned));
|
|
self.exec(
|
|
"INSERT INTO feeds (id, url, title, etag, last_modified, last_checked, ttl_mins, last_error, image, category)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, NULL, $8, $9)
|
|
ON CONFLICT (id) DO UPDATE SET
|
|
url = excluded.url,
|
|
title = coalesce(excluded.title, feeds.title),
|
|
etag = excluded.etag,
|
|
last_modified = excluded.last_modified,
|
|
last_checked = excluded.last_checked,
|
|
ttl_mins = excluded.ttl_mins,
|
|
image = coalesce(excluded.image, feeds.image),
|
|
category = excluded.category,
|
|
last_error = NULL,
|
|
error_since = NULL",
|
|
vec![
|
|
feed_id.into(),
|
|
url.into(),
|
|
o(title),
|
|
o(etag),
|
|
o(last_modified),
|
|
now().into(),
|
|
ttl_mins.map(|t| t as i64).into(),
|
|
o(image),
|
|
o(category),
|
|
],
|
|
)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// 304, or any other poll that produced no new data: only the clock moves.
|
|
pub async fn touch_feed(&self, feed_id: &str, url: &str) -> Result<()> {
|
|
self.exec(
|
|
"INSERT INTO feeds (id, url, last_checked) VALUES ($1, $2, $3)
|
|
ON CONFLICT (id) DO UPDATE SET last_checked = excluded.last_checked,
|
|
last_error = NULL, error_since = NULL",
|
|
vec![feed_id.into(), url.into(), now().into()],
|
|
)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Forgets the cached ETag/Last-Modified. Those validators belong to the old URL, so
|
|
/// keeping them across a URL change could produce a bogus 304 against the new one.
|
|
pub async fn clear_validators(&self, feed_id: &str) -> Result<()> {
|
|
self.exec(
|
|
"UPDATE feeds SET etag = NULL, last_modified = NULL, last_checked = NULL WHERE id = $1",
|
|
vec![feed_id.into()],
|
|
)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
pub async fn set_feed_error(&self, feed_id: &str, url: &str, msg: &str) -> Result<()> {
|
|
self.exec(
|
|
"INSERT INTO feeds (id, url, last_checked, last_error, error_since)
|
|
VALUES ($1, $2, $3, $4, $3)
|
|
ON CONFLICT (id) DO UPDATE SET
|
|
last_checked = excluded.last_checked,
|
|
last_error = excluded.last_error,
|
|
error_since = coalesce(feeds.error_since, excluded.error_since)",
|
|
vec![feed_id.into(), url.into(), now().into(), msg.into()],
|
|
)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Returns true when this entry had not been seen before.
|
|
///
|
|
pub async fn record_entry(&self, feed_id: &str, e: &crate::feed::Entry) -> Result<bool> {
|
|
let inserted = self
|
|
.exec(
|
|
"INSERT INTO entries
|
|
(feed_id, guid, title, link, published, description, first_seen,
|
|
image, duration, episode, season)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
|
|
ON CONFLICT DO NOTHING",
|
|
vec![
|
|
feed_id.into(),
|
|
e.guid.clone().into(),
|
|
e.title.clone().into(),
|
|
e.link.clone().into(),
|
|
e.published.into(),
|
|
e.description.clone().into(),
|
|
now().into(),
|
|
e.image.clone().into(),
|
|
e.duration.into(),
|
|
e.episode.into(),
|
|
e.season.into(),
|
|
],
|
|
)
|
|
.await?;
|
|
if inserted == 0 {
|
|
self.exec(
|
|
"UPDATE entries SET
|
|
title = coalesce($3, title),
|
|
description = coalesce($4, description),
|
|
image = coalesce($5, image),
|
|
duration = coalesce($6, duration),
|
|
episode = coalesce($7, episode),
|
|
season = coalesce($8, season)
|
|
WHERE feed_id = $1 AND guid = $2",
|
|
vec![
|
|
feed_id.into(),
|
|
e.guid.clone().into(),
|
|
e.title.clone().into(),
|
|
e.description.clone().into(),
|
|
e.image.clone().into(),
|
|
e.duration.into(),
|
|
e.episode.into(),
|
|
e.season.into(),
|
|
],
|
|
)
|
|
.await?;
|
|
}
|
|
Ok(inserted == 1)
|
|
}
|
|
|
|
/// Returns true when this enclosure URL is new. False means we have downloaded it
|
|
/// before, or deliberately reaped it -- either way it is not fetched again.
|
|
|
|
pub async fn mark_downloaded(&self, url: &str, path: &std::path::Path, bytes: u64) -> Result<()> {
|
|
self.exec(
|
|
"UPDATE enclosures SET state = 'done', path = $2, bytes_done = $3,
|
|
downloaded_at = $4, last_error = NULL WHERE url = $1",
|
|
vec![url.into(), path.to_string_lossy().into_owned().into(), (bytes as i64).into(), now().into()],
|
|
)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// The row stays -- a failed URL is still a URL we have seen. `state` says why it has
|
|
/// no file, and a retry is an explicit act rather than something a rescan does silently.
|
|
pub async fn mark_enclosure(&self, url: &str, state: &str, error: Option<&str>) -> Result<()> {
|
|
self.exec(
|
|
"UPDATE enclosures SET state = $2, last_error = $3 WHERE url = $1",
|
|
vec![url.into(), state.into(), error.map(str::to_owned).into()],
|
|
)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
pub async fn record_enclosure(
|
|
&self,
|
|
feed_id: &str,
|
|
guid: &str,
|
|
enc: &crate::feed::Enclosure,
|
|
) -> Result<bool> {
|
|
let inserted = self
|
|
.exec(
|
|
"INSERT INTO enclosures (feed_id, guid, url, mime, length, state)
|
|
VALUES ($1, $2, $3, $4, $5, 'pending') ON CONFLICT DO NOTHING",
|
|
vec![
|
|
feed_id.into(),
|
|
guid.into(),
|
|
enc.url.clone().into(),
|
|
enc.mime.clone().into(),
|
|
enc.length.into(),
|
|
],
|
|
)
|
|
.await?;
|
|
Ok(inserted == 1)
|
|
}
|
|
}
|
|
|
|
|
|
/// An enclosure waiting to be downloaded.
|
|
#[derive(Debug)]
|
|
pub struct Pending {
|
|
pub id: i64,
|
|
pub url: String,
|
|
pub mime: Option<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 async fn pending(&self, feed_id: &str, limit: usize) -> Result<Vec<Pending>> {
|
|
// Newest first: a cap of 3 should mean the three latest episodes, not the three that
|
|
// happen to have been recorded first.
|
|
self.rows(
|
|
"SELECT x.id, x.url, x.mime FROM enclosures x
|
|
JOIN entries e ON e.feed_id = x.feed_id AND e.guid = x.guid
|
|
WHERE x.feed_id = $1 AND x.state = 'pending'
|
|
ORDER BY coalesce(e.published, e.first_seen) DESC, x.id DESC
|
|
LIMIT $2",
|
|
vec![feed_id.into(), (limit as i64).into()],
|
|
)
|
|
.await?
|
|
.iter()
|
|
.map(|r| Ok(Pending { id: r.try_get("", "id")?, url: r.try_get("", "url")?, mime: r.try_get("", "mime")? }))
|
|
.collect()
|
|
}
|
|
}
|
|
|
|
|
|
/// A downloaded file the reaper may consider.
|
|
#[derive(Debug, Clone, PartialEq)]
|
|
pub struct Candidate {
|
|
pub id: i64,
|
|
pub url: String,
|
|
pub path: String,
|
|
pub bytes: i64,
|
|
/// When it landed; the reaper works oldest-first.
|
|
pub age_key: i64,
|
|
pub read: bool,
|
|
}
|
|
|
|
impl Db {
|
|
/// Files on disk that may be deleted to get back under quota: starred by nobody,
|
|
/// with the ones everybody has finished going first, oldest first within each group.
|
|
///
|
|
/// One file serves every subscriber, so both tests are about all of them: **anyone**
|
|
/// starring it keeps it, and it only counts as read when **everyone** subscribed has
|
|
/// read it. A file whose feed nobody subscribes to has no one left to keep it, so it
|
|
/// sorts with the read ones.
|
|
///
|
|
/// (The Python intended `read = 1 AND flagged = 0` but never achieved it -- a missing
|
|
/// plistlib import and an `EntreiesData` typo meant the filter always threw.)
|
|
pub async fn reap_candidates(&self) -> Result<Vec<Candidate>> {
|
|
// Yes/no as true and false, not 1 and 0: Postgres types a bare 1 as a 32-bit integer
|
|
// and will not hand it over as an i64.
|
|
self.rows(
|
|
"SELECT e.id, e.url, e.path, e.bytes_done, coalesce(e.downloaded_at, 0) AS age_key,
|
|
CASE WHEN coalesce(readers.n, 0) >= coalesce(subs.n, 0) THEN true ELSE false END AS read
|
|
FROM enclosures e
|
|
LEFT JOIN (SELECT feed_id, count(*) AS n FROM subscriptions GROUP BY feed_id) subs
|
|
ON subs.feed_id = e.feed_id
|
|
LEFT JOIN (SELECT feed_id, guid, count(*) AS n FROM entry_state
|
|
WHERE read GROUP BY feed_id, guid) readers
|
|
ON readers.feed_id = e.feed_id AND readers.guid = e.guid
|
|
WHERE e.path IS NOT NULL
|
|
AND NOT EXISTS (SELECT 1 FROM entry_state s
|
|
WHERE s.feed_id = e.feed_id AND s.guid = e.guid AND s.flagged)
|
|
ORDER BY 6 DESC, coalesce(e.downloaded_at, 0) ASC, e.id ASC",
|
|
vec![],
|
|
)
|
|
.await?
|
|
.iter()
|
|
.map(|r| {
|
|
Ok(Candidate {
|
|
id: r.try_get("", "id")?,
|
|
url: r.try_get("", "url")?,
|
|
path: r.try_get("", "path")?,
|
|
bytes: r.try_get("", "bytes_done")?,
|
|
age_key: r.try_get("", "age_key")?,
|
|
read: r.try_get("", "read")?,
|
|
})
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
/// The row survives the file: that is what stops a reaped episode being re-downloaded.
|
|
pub async fn mark_reaped(&self, id: i64) -> Result<()> {
|
|
self.exec("UPDATE enclosures SET state = 'reaped', path = NULL WHERE id = $1", vec![id.into()]).await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Rows claiming a file that is no longer there (someone deleted it by hand).
|
|
pub async fn missing_files(&self) -> Result<Vec<(i64, String)>> {
|
|
Ok(enclosures::Entity::find()
|
|
.filter(enclosures::Column::Path.is_not_null())
|
|
.all(&self.orm)
|
|
.await?
|
|
.into_iter()
|
|
.filter_map(|e| Some((e.id, e.path?)))
|
|
.filter(|(_, p)| !std::path::Path::new(p).exists())
|
|
.collect())
|
|
}
|
|
|
|
/// Old entries that never had a file, or no longer have one. Enclosure rows stay --
|
|
/// they are the dedupe history.
|
|
pub async fn prune_entries(&self, older_than: i64) -> Result<usize> {
|
|
let n = self
|
|
.exec(
|
|
"DELETE FROM entries
|
|
WHERE coalesce(published, first_seen) < $1
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM enclosures e
|
|
WHERE e.feed_id = entries.feed_id AND e.guid = entries.guid
|
|
AND e.path IS NOT NULL)
|
|
-- Starred by anyone keeps it, the same rule the reaper follows.
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM entry_state s
|
|
WHERE s.feed_id = entries.feed_id AND s.guid = entries.guid
|
|
AND s.flagged)",
|
|
vec![older_than.into()],
|
|
)
|
|
.await?;
|
|
// Whatever went takes everyone's read state with it, rather than leaving rows
|
|
// pointing at an item that no longer exists.
|
|
self.exec(
|
|
"DELETE FROM entry_state WHERE NOT EXISTS (
|
|
SELECT 1 FROM entries e
|
|
WHERE e.feed_id = entry_state.feed_id AND e.guid = entry_state.guid)",
|
|
vec![],
|
|
)
|
|
.await?;
|
|
Ok(n as usize)
|
|
}
|
|
}
|
|
|
|
impl Db {
|
|
pub async fn unread_count(&self, user_id: i64, feed_id: &str) -> Result<i64> {
|
|
let r = self
|
|
.rows(
|
|
"SELECT count(*) AS n FROM entries e
|
|
LEFT JOIN entry_state s
|
|
ON s.user_id = $2 AND s.feed_id = e.feed_id AND s.guid = e.guid
|
|
WHERE e.feed_id = $1 AND NOT coalesce(s.read, false)",
|
|
vec![feed_id.into(), user_id.into()],
|
|
)
|
|
.await?;
|
|
Ok(r.first().context("count(*) always returns a row")?.try_get("", "n")?)
|
|
}
|
|
|
|
/// (pending, downloaded) across all feeds, for the status command.
|
|
pub async fn counts(&self) -> Result<(i64, i64)> {
|
|
let pending = enclosures::Entity::find()
|
|
.filter(enclosures::Column::State.eq("pending"))
|
|
.count(&self.orm)
|
|
.await?;
|
|
let downloaded = enclosures::Entity::find()
|
|
.filter(enclosures::Column::Path.is_not_null())
|
|
.count(&self.orm)
|
|
.await?;
|
|
Ok((pending as i64, downloaded as i64))
|
|
}
|
|
}
|
|
|
|
|
|
/// An entry plus its enclosures, for the web UI.
|
|
#[derive(Debug, serde::Serialize)]
|
|
pub struct EntryRow {
|
|
pub guid: String,
|
|
pub feed_id: String,
|
|
pub title: Option<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>,
|
|
}
|
|
|
|
/// A statement's parameters, gathered as its SQL is written: each `p` binds a value and gives
|
|
/// back its `$n`. Only what the SQL uses is bound -- Postgres refuses a parameter it cannot
|
|
/// place, where rusqlite needed every one mentioned, which is what `?1 IS NULL` was for.
|
|
#[derive(Default)]
|
|
struct Args(Vec<sea_orm::Value>);
|
|
|
|
impl Args {
|
|
fn p(&mut self, v: impl Into<sea_orm::Value>) -> String {
|
|
self.0.push(v.into());
|
|
format!("${}", self.0.len())
|
|
}
|
|
}
|
|
|
|
/// The FROM and WHERE that `entries_in` and `count_in` share: whose read state, which feeds
|
|
/// (one, or every feed the person subscribes to), the filter, and a case-insensitive search of
|
|
/// title and description.
|
|
fn entries_from(a: &mut Args, user_id: i64, feed_id: Option<&str>, filter: Filter, search: Option<&str>) -> String {
|
|
let user = a.p(user_id);
|
|
let scope = match feed_id {
|
|
Some(f) => format!("e.feed_id = {}", a.p(f)),
|
|
None => format!("e.feed_id IN (SELECT feed_id FROM subscriptions WHERE user_id = {user})"),
|
|
};
|
|
let search = match search.map(|q| q.trim().to_lowercase()).filter(|q| !q.is_empty()) {
|
|
Some(q) => {
|
|
let like = a.p(format!("%{q}%"));
|
|
format!(
|
|
"(lower(coalesce(e.title, '')) LIKE {like} OR lower(coalesce(e.description, '')) LIKE {like})"
|
|
)
|
|
}
|
|
None => "true".into(),
|
|
};
|
|
format!(
|
|
"FROM entries e
|
|
LEFT JOIN entry_state s ON s.user_id = {user} AND s.feed_id = e.feed_id AND s.guid = e.guid
|
|
WHERE {scope} AND {} AND {search}",
|
|
filter.sql()
|
|
)
|
|
}
|
|
|
|
/// The item table's ORDER BY. The column name picks one of these fixed expressions, so nothing
|
|
/// the caller sends reaches the query, and anything unrecognised is newest first. Ties fall back
|
|
/// to newest first too, so a page boundary is stable across "Load more".
|
|
///
|
|
/// ponytail: file type and size look at the item's first and largest file. The row shows the file
|
|
/// it summarises, which is almost always that one; sort by that one if they ever disagree.
|
|
/// `pinned_first` puts your pinned items above the rest, each part in the order asked for, so a
|
|
/// pin keeps something at the top of its list (issue #35). Not when sorting by the pin itself,
|
|
/// where the direction chosen is the point.
|
|
pub fn order_sql(col: &str, dir: &str, pinned_first: bool) -> String {
|
|
let expr = match col {
|
|
"kept" => "coalesce(s.flagged, false)",
|
|
"title" => "lower(coalesce(e.title, ''))",
|
|
"feed" => "(SELECT lower(coalesce(f.title, f.id)) FROM feeds f WHERE f.id = e.feed_id)",
|
|
"type" => "(SELECT min(x.mime) FROM enclosures x WHERE x.feed_id = e.feed_id AND x.guid = e.guid)",
|
|
"size" => "(SELECT max(x.length) FROM enclosures x WHERE x.feed_id = e.feed_id AND x.guid = e.guid)",
|
|
_ => "coalesce(e.published, e.first_seen)",
|
|
};
|
|
let dir = if dir == "asc" { "ASC" } else { "DESC" };
|
|
let pins = if pinned_first && col != "kept" { "coalesce(s.flagged, false) DESC, " } else { "" };
|
|
// The guid breaks what ties remain: SQLite's rowid did, and Postgres has none.
|
|
format!("{pins}{expr} {dir}, coalesce(e.published, e.first_seen) DESC, e.guid DESC")
|
|
}
|
|
|
|
/// Which slice of a feed the UI is asking for.
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
|
pub enum Filter {
|
|
All,
|
|
Unread,
|
|
Downloaded,
|
|
Flagged,
|
|
/// Started (a saved playback position past the first few seconds) and short of the 90%
|
|
/// where `markPlayed` in the UI calls it finished. Not `read`: opening an item marks it
|
|
/// read, so filtering on that hid nearly every episode anyone had started. The length is the
|
|
/// one this person's player measured where there is one (`set_position`), else the feed's;
|
|
/// with neither, the episode counts as unfinished.
|
|
/// Currently Listening, below Popular, is this filter on every feed at once.
|
|
InProgress,
|
|
}
|
|
|
|
impl Filter {
|
|
pub fn parse(s: &str) -> Self {
|
|
match s {
|
|
"unread" => Self::Unread,
|
|
"downloaded" => Self::Downloaded,
|
|
"flagged" => Self::Flagged,
|
|
"in_progress" => Self::InProgress,
|
|
_ => Self::All,
|
|
}
|
|
}
|
|
|
|
/// The WHERE fragment for this filter. `e` is entries, and the EXISTS subquery is
|
|
/// correlated against it.
|
|
fn sql(self) -> &'static str {
|
|
match self {
|
|
// true and false, not 1 and 0: the columns are booleans on Postgres, and SQLite
|
|
// reads true and false as 1 and 0.
|
|
Self::All => "true",
|
|
Self::Unread => "NOT coalesce(s.read, false)",
|
|
Self::Flagged => "coalesce(s.flagged, false)",
|
|
Self::Downloaded => {
|
|
"EXISTS (SELECT 1 FROM enclosures x
|
|
WHERE x.feed_id = e.feed_id AND x.guid = e.guid AND x.path IS NOT NULL)"
|
|
}
|
|
Self::InProgress => {
|
|
"coalesce(s.position, 0) > 5
|
|
AND (coalesce(s.duration, e.duration, 0) = 0
|
|
OR s.position * 10 < coalesce(s.duration, e.duration) * 9)"
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
#[derive(Debug, Clone, serde::Serialize)]
|
|
pub struct EncRow {
|
|
pub id: i64,
|
|
pub feed_id: String,
|
|
pub guid: String,
|
|
pub url: String,
|
|
pub mime: Option<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.
|
|
#[allow(clippy::too_many_arguments)]
|
|
pub async fn entries_in(
|
|
&self,
|
|
user_id: i64,
|
|
feed_id: Option<&str>,
|
|
filter: Filter,
|
|
search: Option<&str>,
|
|
offset: i64,
|
|
limit: i64,
|
|
order: &str,
|
|
) -> Result<Vec<EntryRow>> {
|
|
let mut a = Args::default();
|
|
let from = entries_from(&mut a, user_id, feed_id, filter, search);
|
|
let sql = format!(
|
|
"SELECT e.guid, e.feed_id, e.title, e.link, e.published, e.description,
|
|
coalesce(s.read, false) AS read, coalesce(s.flagged, false) AS flagged, e.image,
|
|
coalesce(s.duration, e.duration) AS duration,
|
|
e.episode, e.season, coalesce(s.position, 0) AS position
|
|
{from}
|
|
ORDER BY {order}
|
|
LIMIT {} OFFSET {}",
|
|
a.p(limit),
|
|
a.p(offset)
|
|
);
|
|
let mut rows = self
|
|
.rows(&sql, a.0)
|
|
.await?
|
|
.iter()
|
|
.map(|r| {
|
|
Ok(EntryRow {
|
|
guid: r.try_get("", "guid")?,
|
|
feed_id: r.try_get("", "feed_id")?,
|
|
title: r.try_get("", "title")?,
|
|
link: r.try_get("", "link")?,
|
|
published: r.try_get("", "published")?,
|
|
description: r.try_get("", "description")?,
|
|
read: r.try_get("", "read")?,
|
|
flagged: r.try_get("", "flagged")?,
|
|
image: r.try_get("", "image")?,
|
|
duration: r.try_get("", "duration")?,
|
|
episode: r.try_get("", "episode")?,
|
|
season: r.try_get("", "season")?,
|
|
position: r.try_get("", "position")?,
|
|
enclosures: vec![],
|
|
})
|
|
})
|
|
.collect::<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 encs = enclosures::Entity::find()
|
|
.filter(enclosures::Column::Guid.is_in(rows.iter().map(|r| r.guid.clone())))
|
|
.order_by_asc(enclosures::Column::Id)
|
|
.all(&self.orm)
|
|
.await?;
|
|
for enc in encs {
|
|
if let Some(row) = rows.iter_mut().find(|r| r.guid == enc.guid && r.feed_id == enc.feed_id) {
|
|
row.enclosures.push(EncRow::from(enc));
|
|
}
|
|
}
|
|
Ok(rows)
|
|
}
|
|
|
|
/// How many entries `entries_in` would page through, so the UI knows whether there is more.
|
|
pub async fn count_in(
|
|
&self,
|
|
user_id: i64,
|
|
feed_id: Option<&str>,
|
|
filter: Filter,
|
|
search: Option<&str>,
|
|
) -> Result<i64> {
|
|
let mut a = Args::default();
|
|
let sql = format!("SELECT count(*) AS n {}", entries_from(&mut a, user_id, feed_id, filter, search));
|
|
let r = self.rows(&sql, a.0).await?;
|
|
Ok(r.first().context("count(*) always returns a row")?.try_get("", "n")?)
|
|
}
|
|
|
|
/// Where playback got to, so it resumes there next time -- for this listener only.
|
|
/// `duration` is the length their player measured, kept beside the position and preferred to
|
|
/// the feed's for time left and for when an episode counts as finished. A feed can be minutes
|
|
/// out: ReThinking's gave 41:23 for a 43:48 file, which said 0:08 left with 2:33 to play.
|
|
/// Not written to `entries`, where every scan puts the feed's figure back, and kept per
|
|
/// listener so one person's player never changes what anyone else sees.
|
|
pub async fn set_position(
|
|
&self,
|
|
user_id: i64,
|
|
feed_id: &str,
|
|
guid: &str,
|
|
secs: i64,
|
|
duration: Option<i64>,
|
|
) -> Result<()> {
|
|
// entry_state.duration, named in full: Postgres will not guess between it and excluded's.
|
|
self.exec(
|
|
"INSERT INTO entry_state (user_id, feed_id, guid, position, duration)
|
|
VALUES ($1, $2, $3, $4, $5)
|
|
ON CONFLICT (user_id, feed_id, guid) DO UPDATE SET position = excluded.position,
|
|
duration = coalesce(excluded.duration, entry_state.duration)",
|
|
vec![
|
|
user_id.into(),
|
|
feed_id.into(),
|
|
guid.into(),
|
|
secs.max(0).into(),
|
|
duration.filter(|d| *d > 0).into(),
|
|
],
|
|
)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
// ---- hand-written SQL ----
|
|
//
|
|
// For what reads better as SQL than as a query builder: joins, sums, upserts. Written to
|
|
// run on SQLite and Postgres alike -- $1-style parameters, ON CONFLICT, CASE WHEN on a
|
|
// yes/no column rather than comparing it to 1 -- since both run it (issue #18).
|
|
|
|
fn stmt(&self, sql: &str, values: Vec<sea_orm::Value>) -> Statement {
|
|
Statement::from_sql_and_values(self.orm.get_database_backend(), sql, values)
|
|
}
|
|
|
|
/// Runs a statement; how many rows it changed.
|
|
async fn exec(&self, sql: &str, values: Vec<sea_orm::Value>) -> Result<u64> {
|
|
Ok(self.orm.execute_raw(self.stmt(sql, values)).await?.rows_affected())
|
|
}
|
|
|
|
async fn rows(&self, sql: &str, values: Vec<sea_orm::Value>) -> Result<Vec<sea_orm::QueryResult>> {
|
|
Ok(self.orm.query_all_raw(self.stmt(sql, values)).await?)
|
|
}
|
|
|
|
/// The first admin starts subscribed to the whole catalogue: whoever wrote config.toml meant
|
|
/// to read those feeds, and without this a fresh install signs in to an empty sidebar. Runs
|
|
/// only while nobody subscribes to anything, so an unsubscribe is never undone.
|
|
pub async fn adopt_catalogue(&self, user_id: i64, catalogue: &[String]) -> Result<usize> {
|
|
if subscriptions::Entity::find().count(&self.orm).await? > 0 {
|
|
return Ok(0);
|
|
}
|
|
for id in catalogue {
|
|
self.subscribe(user_id, id).await?;
|
|
}
|
|
// Feeds that exist only in the database (OPML children) count too. SQLite wants the
|
|
// WHERE to tell the SELECT from the ON CONFLICT; Postgres does not mind it.
|
|
self.exec(
|
|
"INSERT INTO subscriptions (user_id, feed_id) SELECT $1, id FROM feeds WHERE true
|
|
ON CONFLICT DO NOTHING",
|
|
vec![user_id.into()],
|
|
)
|
|
.await?;
|
|
Ok(catalogue.len())
|
|
}
|
|
|
|
// ---- subscriptions ----
|
|
|
|
/// What this person wants from a feed. Absent means they do not subscribe at all.
|
|
pub async fn subscription(&self, user_id: i64, feed_id: &str) -> Result<Option<Sub>> {
|
|
Ok(subscriptions::Entity::find_by_id((user_id, feed_id.to_owned()))
|
|
.one(&self.orm)
|
|
.await?
|
|
.map(Sub::from))
|
|
}
|
|
|
|
/// Every feed this person subscribes to, with their settings.
|
|
pub async fn subscriptions_for(&self, user_id: i64) -> Result<Vec<Sub>> {
|
|
Ok(subscriptions::Entity::find()
|
|
.filter(subscriptions::Column::UserId.eq(user_id))
|
|
.all(&self.orm)
|
|
.await?
|
|
.into_iter()
|
|
.map(Sub::from)
|
|
.collect())
|
|
}
|
|
|
|
/// Everyone's settings for one feed. The scanner merges these into what it fetches
|
|
/// and downloads, since one file serves the lot.
|
|
/// In a group, whatever someone has not set on the feed itself comes from their
|
|
/// subscription to the group, as the group's settings dialog has always said it does.
|
|
pub async fn subscribers(&self, feed_id: &str, group: Option<&str>) -> Result<Vec<Sub>> {
|
|
let rows = self
|
|
.rows(
|
|
"SELECT coalesce(c.keywords, p.keywords) AS keywords,
|
|
coalesce(c.auto_download, p.auto_download) AS auto_download,
|
|
coalesce(c.allow_explicit, p.allow_explicit) AS allow_explicit,
|
|
coalesce(c.max_new_per_check, p.max_new_per_check) AS max_new_per_check
|
|
FROM subscriptions c
|
|
LEFT JOIN subscriptions p ON p.user_id = c.user_id AND p.feed_id = $2
|
|
WHERE c.feed_id = $1",
|
|
vec![feed_id.into(), group.map(str::to_owned).into()],
|
|
)
|
|
.await?;
|
|
rows.iter()
|
|
.map(|r| {
|
|
Ok(Sub {
|
|
feed_id: feed_id.to_owned(),
|
|
keywords: keywords(r.try_get("", "keywords")?),
|
|
auto_download: r.try_get("", "auto_download")?,
|
|
allow_explicit: r.try_get("", "allow_explicit")?,
|
|
max_new_per_check: r.try_get("", "max_new_per_check")?,
|
|
})
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
/// Subscribers per feed, for the whole catalogue in one query -- the feed list would
|
|
/// otherwise ask once per feed.
|
|
pub async fn subscriber_counts(&self) -> Result<std::collections::HashMap<String, i64>> {
|
|
self.rows("SELECT feed_id, count(*) AS n FROM subscriptions GROUP BY feed_id", vec![])
|
|
.await?
|
|
.iter()
|
|
.map(|r| Ok((r.try_get("", "feed_id")?, r.try_get("", "n")?)))
|
|
.collect()
|
|
}
|
|
|
|
/// Feeds with any audio or video enclosure: the Directory's Podcasts, with the rest Blogs.
|
|
/// Reaped files keep their rows, so a show whose files have all been purged still counts.
|
|
pub async fn media_feeds(&self) -> Result<std::collections::HashSet<String>> {
|
|
self.rows(
|
|
"SELECT DISTINCT feed_id FROM enclosures WHERE mime LIKE 'audio/%' OR mime LIKE 'video/%'",
|
|
vec![],
|
|
)
|
|
.await?
|
|
.iter()
|
|
.map(|r| Ok(r.try_get("", "feed_id")?))
|
|
.collect()
|
|
}
|
|
|
|
/// Who else would miss this file: subscribers other than `user_id` who have starred
|
|
/// the item or have not read it yet. Deleting is deleting their copy too.
|
|
pub async fn others_wanting(&self, enclosure_id: i64, user_id: i64) -> Result<(i64, i64)> {
|
|
// CASE WHEN on the column itself, not `= 1`: a boolean on Postgres, 0 or 1 on SQLite,
|
|
// and NULL, for someone who never opened the item, falls to the ELSE either way.
|
|
let r = self
|
|
.rows(
|
|
"SELECT sum(CASE WHEN st.flagged THEN 1 ELSE 0 END) AS starred,
|
|
sum(CASE WHEN st.read THEN 0 ELSE 1 END) AS unread
|
|
FROM enclosures e
|
|
JOIN subscriptions s ON s.feed_id = e.feed_id AND s.user_id <> $2
|
|
LEFT JOIN entry_state st
|
|
ON st.user_id = s.user_id AND st.feed_id = e.feed_id AND st.guid = e.guid
|
|
WHERE e.id = $1",
|
|
vec![enclosure_id.into(), user_id.into()],
|
|
)
|
|
.await?;
|
|
let r = r.first().context("an aggregate always returns a row")?;
|
|
Ok((
|
|
r.try_get::<Option<i64>>("", "starred")?.unwrap_or(0),
|
|
r.try_get::<Option<i64>>("", "unread")?.unwrap_or(0),
|
|
))
|
|
}
|
|
|
|
pub async fn subscribe(&self, user_id: i64, feed_id: &str) -> Result<()> {
|
|
self.exec(
|
|
"INSERT INTO subscriptions (user_id, feed_id) VALUES ($1, $2) ON CONFLICT DO NOTHING",
|
|
vec![user_id.into(), feed_id.into()],
|
|
)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// The feeds this person pinned to the top of their list. Kept apart from `Sub`, which is
|
|
/// what the scanner merges into its policy, and which a pin has nothing to do with.
|
|
pub async fn pinned_feeds(&self, user_id: i64) -> Result<std::collections::HashSet<String>> {
|
|
Ok(subscriptions::Entity::find()
|
|
.filter(subscriptions::Column::UserId.eq(user_id))
|
|
.filter(subscriptions::Column::Pinned.eq(true))
|
|
.all(&self.orm)
|
|
.await?
|
|
.into_iter()
|
|
.map(|s| s.feed_id)
|
|
.collect())
|
|
}
|
|
|
|
/// False when they do not subscribe to it, since there is then no row in their list to pin.
|
|
pub async fn set_pinned(&self, user_id: i64, feed_id: &str, on: bool) -> Result<bool> {
|
|
let r = subscriptions::Entity::update_many()
|
|
.col_expr(subscriptions::Column::Pinned, Expr::val(on).into())
|
|
.filter(subscriptions::Column::UserId.eq(user_id))
|
|
.filter(subscriptions::Column::FeedId.eq(feed_id))
|
|
.exec(&self.orm)
|
|
.await?;
|
|
Ok(r.rows_affected > 0)
|
|
}
|
|
|
|
pub async fn unsubscribe(&self, user_id: i64, feed_id: &str) -> Result<()> {
|
|
subscriptions::Entity::delete_by_id((user_id, feed_id.to_owned())).exec(&self.orm).await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Overwrites one person's settings for a feed. A None field means: follow the feed.
|
|
/// Names its columns, so the pin, which is not a setting, is left as it was.
|
|
pub async fn set_subscription(&self, user_id: i64, sub: &Sub) -> Result<()> {
|
|
let kw = sub.keywords.as_ref().map(serde_json::to_string).transpose()?;
|
|
self.exec(
|
|
"INSERT INTO subscriptions
|
|
(user_id, feed_id, keywords, auto_download, allow_explicit, max_new_per_check)
|
|
VALUES ($1, $2, $3, $4, $5, $6)
|
|
ON CONFLICT (user_id, feed_id) DO UPDATE SET
|
|
keywords = excluded.keywords,
|
|
auto_download = excluded.auto_download,
|
|
allow_explicit = excluded.allow_explicit,
|
|
max_new_per_check = excluded.max_new_per_check",
|
|
vec![
|
|
user_id.into(),
|
|
sub.feed_id.clone().into(),
|
|
kw.into(),
|
|
sub.auto_download.into(),
|
|
sub.allow_explicit.into(),
|
|
sub.max_new_per_check.into(),
|
|
],
|
|
)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
// ---- users and sessions ----
|
|
|
|
pub async fn create_user(&self, name: &str, pass_hash: Option<&str>, admin: bool) -> Result<i64> {
|
|
let m = users::ActiveModel {
|
|
name: Set(name.to_owned()),
|
|
pass_hash: Set(pass_hash.map(str::to_owned)),
|
|
is_admin: Set(admin),
|
|
created: Set(Some(now())),
|
|
..Default::default()
|
|
}
|
|
.insert(&self.orm)
|
|
.await?;
|
|
Ok(m.id)
|
|
}
|
|
|
|
/// Without regard to case, on either database: lower() both sides, which the unique index
|
|
/// on lower(name) serves. SQLite's COLLATE NOCASE on the column did this before, and
|
|
/// Postgres has no such thing.
|
|
pub async fn user_by_name(&self, name: &str) -> Result<Option<User>> {
|
|
use sea_orm::sea_query::ExprTrait;
|
|
Ok(users::Entity::find()
|
|
.filter(Expr::expr(Func::lower(Expr::col(users::Column::Name))).eq(Func::lower(name)))
|
|
.one(&self.orm)
|
|
.await?
|
|
.map(User::from))
|
|
}
|
|
|
|
pub async fn user_by_id(&self, id: i64) -> Result<Option<User>> {
|
|
Ok(users::Entity::find_by_id(id).one(&self.orm).await?.map(User::from))
|
|
}
|
|
|
|
pub async fn users(&self) -> Result<Vec<User>> {
|
|
Ok(users::Entity::find()
|
|
.order_by_asc(users::Column::Name)
|
|
.all(&self.orm)
|
|
.await?
|
|
.into_iter()
|
|
.map(User::from)
|
|
.collect())
|
|
}
|
|
|
|
/// Sets some of one person's columns, whatever else is in their row.
|
|
async fn update_user(&self, id: i64, m: users::ActiveModel) -> Result<()> {
|
|
users::Entity::update_many()
|
|
.set(m)
|
|
.filter(users::Column::Id.eq(id))
|
|
.exec(&self.orm)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
pub async fn set_password(&self, id: i64, hash: &str) -> Result<()> {
|
|
self.update_user(id, users::ActiveModel { pass_hash: Set(Some(hash.to_owned())), ..Default::default() })
|
|
.await
|
|
}
|
|
|
|
pub async fn set_admin(&self, id: i64, admin: bool) -> Result<()> {
|
|
self.update_user(id, users::ActiveModel { is_admin: Set(admin), ..Default::default() }).await
|
|
}
|
|
|
|
/// The proxy signs people in by the name it vouches for, so an account made before the proxy
|
|
/// was set up has to take that name to be found by it. The name is unique, so a taken one is
|
|
/// refused here as well as by the caller.
|
|
pub async fn rename_user(&self, id: i64, name: &str) -> Result<()> {
|
|
self.update_user(id, users::ActiveModel { name: Set(name.to_owned()), ..Default::default() }).await
|
|
}
|
|
|
|
/// Records a sign-in, to the hour: the proxy vouches for every request, and writing each one
|
|
/// would buy nothing.
|
|
pub async fn signed_in(&self, id: i64) -> Result<()> {
|
|
// Here only: it is implemented for every type, and file-wide it shadows i64::max.
|
|
use sea_orm::sea_query::ExprTrait;
|
|
let now = now();
|
|
users::Entity::update_many()
|
|
.col_expr(users::Column::LastLogin, Expr::val(now).into())
|
|
.filter(users::Column::Id.eq(id))
|
|
.filter(Expr::expr(Func::coalesce([Expr::col(users::Column::LastLogin), Expr::val(0)])).lte(now - 3600))
|
|
.exec(&self.orm)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Sessions go with the user: a deleted account must not leave a usable cookie behind.
|
|
/// The foreign key would take them anyway; this does not rely on it being switched on.
|
|
pub async fn delete_user(&self, id: i64) -> Result<()> {
|
|
sessions::Entity::delete_many().filter(sessions::Column::UserId.eq(id)).exec(&self.orm).await?;
|
|
users::Entity::delete_by_id(id).exec(&self.orm).await?;
|
|
Ok(())
|
|
}
|
|
|
|
pub async fn create_session(&self, user_id: i64, token: &str) -> Result<()> {
|
|
sessions::ActiveModel { token: Set(token.to_owned()), user_id: Set(user_id), seen: Set(now()) }
|
|
.insert(&self.orm)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// The theme this person chose, and light, dark or auto; None for either until they choose.
|
|
pub async fn theme(&self, user_id: i64) -> Result<(Option<String>, Option<String>)> {
|
|
Ok(users::Entity::find_by_id(user_id)
|
|
.one(&self.orm)
|
|
.await?
|
|
.map(|u| (u.theme, u.theme_mode))
|
|
.unwrap_or_default())
|
|
}
|
|
|
|
pub async fn set_theme(&self, user_id: i64, theme: &str, mode: &str) -> Result<()> {
|
|
self.update_user(
|
|
user_id,
|
|
users::ActiveModel {
|
|
theme: Set(Some(theme.to_owned())),
|
|
theme_mode: Set(Some(mode.to_owned())),
|
|
..Default::default()
|
|
},
|
|
)
|
|
.await
|
|
}
|
|
|
|
/// The user behind a session cookie, if it is still live. Idle sessions expire after
|
|
/// `max_idle_secs`; touching `seen` is what keeps a session in daily use alive.
|
|
pub async fn session_user(&self, token: &str, max_idle_secs: i64) -> Result<Option<User>> {
|
|
// Here only: it is implemented for every type, and file-wide it shadows i64::max.
|
|
use sea_orm::sea_query::ExprTrait;
|
|
let cutoff = now() - max_idle_secs;
|
|
let found = sessions::Entity::find_by_id(token.to_owned())
|
|
.filter(sessions::Column::Seen.gte(cutoff))
|
|
.find_also_related(users::Entity)
|
|
.one(&self.orm)
|
|
.await?
|
|
.and_then(|(_, u)| u);
|
|
if found.is_some() {
|
|
sessions::Entity::update_many()
|
|
.col_expr(sessions::Column::Seen, Expr::val(now()).into())
|
|
.filter(sessions::Column::Token.eq(token))
|
|
.exec(&self.orm)
|
|
.await?;
|
|
} else {
|
|
// Either unknown or timed out; either way it is dead weight.
|
|
sessions::Entity::delete_many()
|
|
.filter(sessions::Column::Token.eq(token).or(sessions::Column::Seen.lt(cutoff)))
|
|
.exec(&self.orm)
|
|
.await?;
|
|
}
|
|
Ok(found.map(User::from))
|
|
}
|
|
|
|
pub async fn delete_session(&self, token: &str) -> Result<()> {
|
|
sessions::Entity::delete_by_id(token.to_owned()).exec(&self.orm).await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Marks every entry of the given feeds read. Takes a list because an OPML subscription
|
|
/// holds no entries itself -- marking it read means the feeds inside it.
|
|
pub async fn mark_all_read(&self, user_id: i64, feed_ids: &[String]) -> Result<usize> {
|
|
let mut n = 0;
|
|
for id in feed_ids {
|
|
n += self
|
|
.exec(
|
|
"INSERT INTO entry_state (user_id, feed_id, guid, read)
|
|
SELECT $1, e.feed_id, e.guid, true FROM entries e
|
|
LEFT JOIN entry_state s
|
|
ON s.user_id = $1 AND s.feed_id = e.feed_id AND s.guid = e.guid
|
|
WHERE e.feed_id = $2 AND NOT coalesce(s.read, false)
|
|
ON CONFLICT (user_id, feed_id, guid) DO UPDATE SET read = true",
|
|
vec![user_id.into(), id.clone().into()],
|
|
)
|
|
.await? as usize;
|
|
}
|
|
Ok(n)
|
|
}
|
|
|
|
/// The next N enclosures with no file, newest entry first -- what "download latest"
|
|
/// queues up.
|
|
pub async fn undownloaded(&self, feed_id: &str, limit: i64) -> Result<Vec<i64>> {
|
|
self.rows(
|
|
"SELECT x.id FROM enclosures x
|
|
JOIN entries e ON e.feed_id = x.feed_id AND e.guid = x.guid
|
|
WHERE x.feed_id = $1 AND x.path IS NULL AND x.state <> 'reaped'
|
|
ORDER BY coalesce(e.published, e.first_seen) DESC
|
|
LIMIT $2",
|
|
vec![feed_id.into(), limit.into()],
|
|
)
|
|
.await?
|
|
.iter()
|
|
.map(|r| Ok(r.try_get("", "id")?))
|
|
.collect()
|
|
}
|
|
|
|
pub async fn enclosure(&self, id: i64) -> Result<Option<EncRow>> {
|
|
Ok(enclosures::Entity::find_by_id(id).one(&self.orm).await?.map(EncRow::from))
|
|
}
|
|
|
|
/// Read and kept, per person. The row is created on first touch.
|
|
pub async fn set_entry_flag(
|
|
&self,
|
|
user_id: i64,
|
|
feed_id: &str,
|
|
guid: &str,
|
|
field: EntryFlag,
|
|
on: bool,
|
|
) -> Result<()> {
|
|
let col = match field {
|
|
EntryFlag::Read => "read",
|
|
EntryFlag::Flagged => "flagged",
|
|
};
|
|
self.exec(
|
|
&format!(
|
|
"INSERT INTO entry_state (user_id, feed_id, guid, {col})
|
|
VALUES ($1, $2, $3, $4)
|
|
ON CONFLICT (user_id, feed_id, guid) DO UPDATE SET {col} = excluded.{col}"
|
|
),
|
|
vec![user_id.into(), feed_id.into(), guid.into(), on.into()],
|
|
)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// How many files this feed has on disk. Decides whether a feed dropped from an OPML
|
|
/// can be removed or must be kept.
|
|
pub async fn downloaded_count(&self, feed_id: &str) -> Result<i64> {
|
|
Ok(enclosures::Entity::find()
|
|
.filter(enclosures::Column::FeedId.eq(feed_id))
|
|
.filter(enclosures::Column::Path.is_not_null())
|
|
.count(&self.orm)
|
|
.await? as i64)
|
|
}
|
|
|
|
/// Names a feed without touching its conditional-GET validators. An OPML subscription
|
|
/// takes its name from the document's own <head><title>.
|
|
pub async fn set_title(&self, feed_id: &str, title: &str) -> Result<()> {
|
|
self.exec(
|
|
"UPDATE feeds SET title = $2 WHERE id = $1 AND coalesce(title, '') <> $2",
|
|
vec![feed_id.into(), title.into()],
|
|
)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Records a feed that came from an OPML. Its settings are the parent's; only what
|
|
/// identifies it is stored.
|
|
pub async fn upsert_managed(&self, id: &str, url: &str, title: &str, group_id: &str) -> Result<()> {
|
|
self.exec(
|
|
"INSERT INTO feeds (id, url, title, group_id, managed, orphaned)
|
|
VALUES ($1, $2, $3, $4, true, false)
|
|
ON CONFLICT (id) DO UPDATE SET
|
|
url = excluded.url,
|
|
title = coalesce(feeds.title, excluded.title),
|
|
group_id = excluded.group_id,
|
|
managed = true,
|
|
orphaned = false",
|
|
vec![id.into(), url.into(), title.into(), group_id.into()],
|
|
)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Every feed derived from an OPML, whichever group.
|
|
pub async fn managed_feeds(&self) -> Result<Vec<Managed>> {
|
|
self.rows(
|
|
"SELECT id, url, title, group_id FROM feeds
|
|
WHERE managed AND group_id IS NOT NULL ORDER BY coalesce(title, id)",
|
|
vec![],
|
|
)
|
|
.await?
|
|
.iter()
|
|
.map(|r| {
|
|
Ok(Managed {
|
|
id: r.try_get("", "id")?,
|
|
url: r.try_get("", "url")?,
|
|
title: r.try_get("", "title")?,
|
|
group_id: r.try_get("", "group_id")?,
|
|
})
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
/// Forgets a derived feed entirely. Only for one with nothing downloaded.
|
|
pub async fn drop_managed(&self, id: &str) -> Result<()> {
|
|
self.exec("DELETE FROM feeds WHERE id = $1 AND managed", vec![id.into()]).await?;
|
|
self.exec("DELETE FROM entries WHERE feed_id = $1", vec![id.into()]).await?;
|
|
self.exec("DELETE FROM enclosures WHERE feed_id = $1 AND path IS NULL", vec![id.into()]).await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Folds enclosures of one item that `key` says are the same file into the first of them, for
|
|
/// WordPress's numbered player URLs (`feed::same_file_key`). The first is the one the parser
|
|
/// keeps, so it keeps its row, taking a repeat's file if it has none of its own; the repeats'
|
|
/// rows go. Returns how many went and the copies left spare, for the caller to delete.
|
|
pub async fn merge_repeated_enclosures(&self, key: impl Fn(&str) -> String) -> Result<(usize, Vec<String>)> {
|
|
use sea_orm::TransactionTrait;
|
|
use std::collections::hash_map::Entry;
|
|
let backend = self.orm.get_database_backend();
|
|
let tx = self.orm.begin().await?;
|
|
// Items with a URL carrying WordPress's `_=` parameter. A LIKE, with the underscore
|
|
// escaped, where SQLite had GLOB '*[?&]_=[0-9]*', which Postgres lacks. It lets through
|
|
// `_=` without a number too, which is harmless: `key` only folds `_=` and digits.
|
|
let rows = tx
|
|
.query_all_raw(Statement::from_string(
|
|
backend,
|
|
r"SELECT id, feed_id, guid, url, path FROM enclosures
|
|
WHERE (feed_id, guid) IN
|
|
(SELECT feed_id, guid FROM enclosures
|
|
WHERE url LIKE '%?\_=%' ESCAPE '\' OR url LIKE '%&\_=%' ESCAPE '\')
|
|
ORDER BY id",
|
|
))
|
|
.await?;
|
|
// The first row of each file, and whether it has the file on disk yet.
|
|
let mut first: std::collections::HashMap<(String, String, String), (i64, bool)> = Default::default();
|
|
let (mut gone, mut spare) = (0, vec![]);
|
|
for r in rows {
|
|
let (id, feed, guid, url, path): (i64, String, String, String, Option<String>) = (
|
|
r.try_get("", "id")?,
|
|
r.try_get("", "feed_id")?,
|
|
r.try_get("", "guid")?,
|
|
r.try_get("", "url")?,
|
|
r.try_get("", "path")?,
|
|
);
|
|
match first.entry((feed, guid, key(&url))) {
|
|
Entry::Vacant(v) => {
|
|
v.insert((id, path.is_some()));
|
|
}
|
|
Entry::Occupied(mut o) => {
|
|
let (keep, has) = o.get_mut();
|
|
if let Some(p) = path {
|
|
if *has {
|
|
spare.push(p);
|
|
} else {
|
|
// The only copy is the repeat's: Rands' episode 97 was downloaded
|
|
// under its ?_=2 URL alone.
|
|
tx.execute_raw(Statement::from_sql_and_values(
|
|
backend,
|
|
"UPDATE enclosures SET (path, state, bytes_done, downloaded_at) =
|
|
(SELECT path, state, bytes_done, downloaded_at FROM enclosures WHERE id = $2)
|
|
WHERE id = $1",
|
|
vec![(*keep).into(), id.into()],
|
|
))
|
|
.await?;
|
|
*has = true;
|
|
}
|
|
}
|
|
tx.execute_raw(Statement::from_sql_and_values(
|
|
backend,
|
|
"DELETE FROM enclosures WHERE id = $1",
|
|
vec![id.into()],
|
|
))
|
|
.await?;
|
|
gone += 1;
|
|
}
|
|
}
|
|
}
|
|
tx.commit().await?;
|
|
Ok((gone, spare))
|
|
}
|
|
|
|
/// Stops treating a feed as derived, because it now has its own config entry.
|
|
pub async fn unmanage(&self, id: &str) -> Result<()> {
|
|
self.exec("UPDATE feeds SET managed = false WHERE id = $1", vec![id.into()]).await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Empties a feed of its items, leaving its files alone.
|
|
pub async fn clear_entries(&self, feed_id: &str) -> Result<()> {
|
|
entries::Entity::delete_many().filter(entries::Column::FeedId.eq(feed_id)).exec(&self.orm).await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Every feed the database holds rows for, as (id, url), removed ones included.
|
|
pub async fn feed_urls(&self) -> Result<Vec<(String, String)>> {
|
|
Ok(feeds::Entity::find().all(&self.orm).await?.into_iter().map(|f| (f.id, f.url)).collect())
|
|
}
|
|
|
|
/// Hands a feed in a group the enclosures its parent holds, as (guid, url), with everyone's
|
|
/// read state for them. A Patreon creator read as one feed before it was split into shows
|
|
/// owns every show's files, and `enclosures.url` is unique, so without this each show would
|
|
/// list its items with nothing to play.
|
|
pub async fn adopt(&self, parent: &str, child: &str, listed: &[(&str, &str)]) -> Result<()> {
|
|
use sea_orm::TransactionTrait;
|
|
let holds = enclosures::Entity::find()
|
|
.filter(enclosures::Column::FeedId.eq(parent))
|
|
.count(&self.orm)
|
|
.await?
|
|
> 0;
|
|
if !holds {
|
|
return Ok(()); // An OPML, or a creator already shared out.
|
|
}
|
|
let backend = self.orm.get_database_backend();
|
|
let tx = self.orm.begin().await?;
|
|
for &(guid, url) in listed {
|
|
let moved = tx
|
|
.execute_raw(Statement::from_sql_and_values(
|
|
backend,
|
|
"UPDATE enclosures SET feed_id = $3, guid = $4 WHERE url = $1 AND feed_id = $2",
|
|
vec![url.into(), parent.into(), child.into(), guid.into()],
|
|
))
|
|
.await?
|
|
.rows_affected();
|
|
if moved == 1 {
|
|
// Patreon gives a post the same guid in every feed it appears in. Someone who
|
|
// already has a row for it under the child keeps that one; SQLite's UPDATE OR
|
|
// IGNORE did this, and Postgres has no such thing.
|
|
tx.execute_raw(Statement::from_sql_and_values(
|
|
backend,
|
|
"UPDATE entry_state SET feed_id = $2 WHERE feed_id = $1 AND guid = $3
|
|
AND NOT EXISTS (SELECT 1 FROM entry_state t
|
|
WHERE t.user_id = entry_state.user_id
|
|
AND t.feed_id = $2 AND t.guid = $3)",
|
|
vec![parent.into(), child.into(), guid.into()],
|
|
))
|
|
.await?;
|
|
}
|
|
}
|
|
tx.commit().await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// A feed's enclosures skipped by one of its filters, by URL, with the reason: the verdicts a
|
|
/// change of settings can overturn. A torrent held back while torrents are off is not a
|
|
/// filter's call.
|
|
pub async fn skipped_by_filter(&self, feed_id: &str) -> Result<std::collections::HashMap<String, String>> {
|
|
self.rows(
|
|
"SELECT url, last_error FROM enclosures
|
|
WHERE feed_id = $1 AND state = 'skipped' AND last_error IS NOT NULL
|
|
AND last_error <> 'torrents disabled'",
|
|
vec![feed_id.into()],
|
|
)
|
|
.await?
|
|
.iter()
|
|
.map(|r| Ok((r.try_get("", "url")?, r.try_get("", "last_error")?)))
|
|
.collect()
|
|
}
|
|
|
|
pub async fn set_orphaned(&self, feed_id: &str, on: bool) -> Result<()> {
|
|
self.exec(
|
|
"INSERT INTO feeds (id, url, orphaned) VALUES ($1, '', $2)
|
|
ON CONFLICT (id) DO UPDATE SET orphaned = excluded.orphaned",
|
|
vec![feed_id.into(), on.into()],
|
|
)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Nothing can be in flight the moment the daemon starts, so any row still marked
|
|
/// `downloading` is a leftover from a restart or a crash. Left alone it would sit
|
|
/// there forever: the pending queue skips it and nothing else ever revisits it.
|
|
pub async fn requeue_interrupted(&self) -> Result<usize> {
|
|
Ok(self
|
|
.exec("UPDATE enclosures SET state = 'pending' WHERE state = 'downloading' AND path IS NULL", vec![])
|
|
.await? as usize)
|
|
}
|
|
|
|
/// Puts an enclosure back in the queue so the next scan picks it up. This is how a
|
|
/// `skipped` verdict (from a filter that has since been changed) gets revisited.
|
|
pub async fn requeue(&self, id: i64) -> Result<()> {
|
|
self.exec(
|
|
"UPDATE enclosures SET state = 'pending', last_error = NULL WHERE id = $1 AND path IS NULL",
|
|
vec![id.into()],
|
|
)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
#[derive(Debug, Clone, Copy)]
|
|
pub enum EntryFlag {
|
|
Read,
|
|
Flagged,
|
|
}
|
|
|
|
/// Unix seconds. Everything time-shaped in the DB is stored this way.
|
|
pub fn now() -> i64 {
|
|
std::time::SystemTime::now()
|
|
.duration_since(std::time::UNIX_EPOCH)
|
|
.map(|d| d.as_secs() as i64)
|
|
.unwrap_or(0)
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
#[tokio::test]
|
|
async fn a_file_wordpress_listed_twice_is_folded_into_one() {
|
|
let db = Db::memory().await.unwrap();
|
|
db.exec_for_test(
|
|
"INSERT INTO enclosures (id, feed_id, guid, url, path, state) VALUES
|
|
(1,'f','a','https://x/a.mp3','/d/a-2.mp3','done'),
|
|
(2,'f','a','https://x/a.mp3?_=2','/d/a.mp3','done'),
|
|
(3,'f','b','https://x/b.mp3',NULL,'reaped'),
|
|
(4,'f','b','https://x/b.mp3?_=2','/d/b.mp3','done'),
|
|
(5,'f','c','https://x/c.mp3?_=1','/d/c.mp3','done'),
|
|
(6,'f','d','https://x/d1.mp3?_=1','/d/d1.mp3','done'),
|
|
(7,'f','d','https://x/d2.mp3?_=2','/d/d2.mp3','done');",
|
|
).await
|
|
.unwrap();
|
|
let key = crate::feed::same_file_key;
|
|
assert_eq!(db.merge_repeated_enclosures(key).await.unwrap(), (2, vec!["/d/a.mp3".to_string()]));
|
|
assert_eq!(
|
|
db.i64s_for_test("SELECT id FROM enclosures ORDER BY id").await,
|
|
[1, 3, 5, 6, 7],
|
|
"a lone ?_=1 and two different files stay"
|
|
);
|
|
assert_eq!(
|
|
db.strings_for_test("SELECT path FROM enclosures WHERE id = 3 UNION ALL SELECT state FROM enclosures WHERE id = 3")
|
|
.await,
|
|
["/d/b.mp3", "done"],
|
|
"the only copy moves, not deleted"
|
|
);
|
|
assert_eq!(db.merge_repeated_enclosures(key).await.unwrap(), (0, vec![]), "and only once");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn every_sort_column_runs_and_orders_both_ways() {
|
|
let db = Db::memory().await.unwrap();
|
|
db.exec_for_test(
|
|
"INSERT INTO users (id, name, is_admin) VALUES (1,'ray',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');",
|
|
).await
|
|
.unwrap();
|
|
let order = async |col: &str, dir: &str| -> Vec<String> {
|
|
db.entries_in(1, None, Filter::All, None, 0, 50, &order_sql(col, dir, false)).await
|
|
.unwrap()
|
|
.into_iter()
|
|
.map(|e| e.guid)
|
|
.collect()
|
|
};
|
|
assert_eq!(order("title", "asc").await, ["b", "a", "c"], "Apple, banana, cherry: case folded");
|
|
assert_eq!(order("title", "desc").await, ["c", "a", "b"]);
|
|
assert_eq!(order("feed", "asc").await, ["b", "c", "a"], "Aardvark, then Zebra's newest first");
|
|
assert_eq!(order("type", "asc").await, ["a", "b", "c"], "audio, image, video");
|
|
assert_eq!(order("size", "desc").await, ["c", "a", "b"]);
|
|
assert_eq!(order("published", "desc").await, ["c", "b", "a"]);
|
|
db.set_entry_flag(1, "f", "a", EntryFlag::Flagged, true).await.unwrap();
|
|
assert_eq!(order("kept", "desc").await[0], "a");
|
|
// Pinned first: the pinned banana tops every sort, the rest in the order asked for.
|
|
let pinned = async |col: &str, dir: &str| -> Vec<String> {
|
|
db.entries_in(1, None, Filter::All, None, 0, 50, &order_sql(col, dir, true)).await
|
|
.unwrap()
|
|
.into_iter()
|
|
.map(|e| e.guid)
|
|
.collect()
|
|
};
|
|
assert_eq!(pinned("published", "desc").await, ["a", "c", "b"]);
|
|
assert_eq!(pinned("title", "desc").await, ["a", "c", "b"]);
|
|
assert_eq!(pinned("kept", "asc").await[2], "a", "sorting by the pin itself keeps its direction");
|
|
// An unknown column or direction is newest first; the name itself never reaches the SQL.
|
|
assert_eq!(order("title; DROP TABLE entries", "sideways").await, ["c", "b", "a"]);
|
|
assert!(!order_sql("x'; --", "asc", false).contains("x'"));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn deleting_a_shared_file_asks_about_everyone_else() {
|
|
let db = Db::memory().await.unwrap();
|
|
db.exec_for_test(
|
|
"INSERT INTO users (id, name, is_admin) VALUES (1,'ray',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');",
|
|
).await
|
|
.unwrap();
|
|
|
|
// Nobody has touched it: both others still have it unplayed.
|
|
assert_eq!(db.others_wanting(1, 1).await.unwrap(), (0, 2));
|
|
|
|
// Sam reads it, Kit stars it.
|
|
db.set_entry_flag(2, "f", "a", EntryFlag::Read, true).await.unwrap();
|
|
db.set_entry_flag(3, "f", "a", EntryFlag::Flagged, true).await.unwrap();
|
|
assert_eq!(db.others_wanting(1, 1).await.unwrap(), (1, 1), "one starred it, one has not played it");
|
|
|
|
// Asking as Kit, only Ray and Sam count -- and Kit's own star is not a reason to
|
|
// warn Kit.
|
|
db.set_entry_flag(1, "f", "a", EntryFlag::Read, true).await.unwrap();
|
|
assert_eq!(db.others_wanting(1, 3).await.unwrap(), (0, 0));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn read_state_belongs_to_one_person() {
|
|
let db = Db::memory().await.unwrap();
|
|
db.exec_for_test(
|
|
"INSERT INTO users (id, name, is_admin) VALUES (1,'ray',1),(2,'sam',0);
|
|
INSERT INTO entries (feed_id, guid, title, first_seen) VALUES
|
|
('f','a','One',100),('f','b','Two',200);",
|
|
).await
|
|
.unwrap();
|
|
|
|
assert_eq!(db.unread_count(1, "f").await.unwrap(), 2);
|
|
assert_eq!(db.unread_count(2, "f").await.unwrap(), 2);
|
|
|
|
db.set_entry_flag(1, "f", "a", EntryFlag::Read, true).await.unwrap();
|
|
assert_eq!(db.unread_count(1, "f").await.unwrap(), 1, "ray read one of them");
|
|
assert_eq!(db.unread_count(2, "f").await.unwrap(), 2, "sam has read nothing");
|
|
|
|
// Starring and position are just as private.
|
|
db.set_entry_flag(1, "f", "b", EntryFlag::Flagged, true).await.unwrap();
|
|
db.set_position(2, "f", "b", 42, Some(600)).await.unwrap();
|
|
let order = order_sql("published", "desc", false);
|
|
let page = async |user| db.entries_in(user, Some("f"), Filter::All, None, 0, 50, &order).await.unwrap();
|
|
let (ray, sam) = (page(1).await, page(2).await);
|
|
let ray_b = ray.iter().find(|e| e.guid == "b").unwrap();
|
|
let sam_b = sam.iter().find(|e| e.guid == "b").unwrap();
|
|
assert!(ray_b.flagged && ray_b.position == 0);
|
|
assert!(!sam_b.flagged && sam_b.position == 42);
|
|
// So is the length sam's player measured.
|
|
assert_ne!(ray_b.duration, Some(600));
|
|
assert_eq!(sam_b.duration, Some(600));
|
|
|
|
// Marking a whole feed read is likewise one person's business.
|
|
assert_eq!(db.mark_all_read(2, &["f".to_string()]).await.unwrap(), 2);
|
|
assert_eq!(db.unread_count(2, "f").await.unwrap(), 0);
|
|
assert_eq!(db.unread_count(1, "f").await.unwrap(), 1);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn schema_is_idempotent_and_summary_handles_unknown_feeds() {
|
|
let db = Db::memory().await.unwrap();
|
|
// Creating what is missing again must not fail: open() does it on every start.
|
|
create_missing(&db.orm).await.unwrap();
|
|
|
|
let sum = db.feed_summary("never-seen").await.unwrap();
|
|
assert_eq!(sum.last_checked, None);
|
|
assert_eq!(sum.entries, 0);
|
|
assert_eq!(sum.downloaded, 0);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_renamed_account_keeps_everything_but_its_name() {
|
|
let db = Db::memory().await.unwrap();
|
|
let ray = db.create_user("rays", None, true).await.unwrap();
|
|
db.create_user("sam", None, false).await.unwrap();
|
|
db.subscribe(ray, "f").await.unwrap();
|
|
db.rename_user(ray, "rays@sdf1.net").await.unwrap();
|
|
assert!(db.user_by_name("rays").await.unwrap().is_none());
|
|
let renamed = db.user_by_name("RAYS@sdf1.net").await.unwrap().unwrap();
|
|
assert_eq!((renamed.id, renamed.is_admin), (ray, true), "same account, still the admin");
|
|
assert_eq!(db.subscriptions_for(ray).await.unwrap().len(), 1, "and still subscribed");
|
|
assert!(db.rename_user(ray, "sam").await.is_err(), "a taken name is refused");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn an_account_knows_when_it_was_made_and_last_signed_in() {
|
|
let db = Db::memory().await.unwrap();
|
|
let id = db.create_user("ray", None, true).await.unwrap();
|
|
let get = async || db.user_by_id(id).await.unwrap().unwrap();
|
|
assert!(get().await.created.is_some_and(|t| t > 0));
|
|
assert_eq!(get().await.last_login, None, "made, but never signed in");
|
|
db.signed_in(id).await.unwrap();
|
|
let first = get().await.last_login.unwrap();
|
|
// Within the hour, the proxy vouching again writes nothing; after it, it does.
|
|
db.exec_for_test(&format!("UPDATE users SET last_login = {} WHERE id = {id}", first - 60)).await.unwrap();
|
|
db.signed_in(id).await.unwrap();
|
|
assert_eq!(get().await.last_login, Some(first - 60));
|
|
db.exec_for_test(&format!("UPDATE users SET last_login = {} WHERE id = {id}", first - 7200)).await.unwrap();
|
|
db.signed_in(id).await.unwrap();
|
|
assert!(get().await.last_login.unwrap() >= first);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn the_first_admin_starts_with_the_catalogue_and_only_once() {
|
|
// Cutting this along with the dead read columns left the browser suite's admin with an
|
|
// empty sidebar: it is how a fresh install's first account gets config.toml's feeds.
|
|
let db = Db::memory().await.unwrap();
|
|
db.exec_for_test("INSERT INTO users (id, name, is_admin) VALUES (1,'admin',1);").await
|
|
.unwrap();
|
|
let subs = async || db.i64s_for_test("SELECT count(*) FROM subscriptions").await[0];
|
|
let catalogue = ["a".to_string(), "b".to_string()];
|
|
assert_eq!(db.adopt_catalogue(1, &catalogue).await.unwrap(), 2);
|
|
assert_eq!(subs().await, 2);
|
|
// Once anyone subscribes to anything it never runs again, so an unsubscribe sticks.
|
|
db.unsubscribe(1, "a").await.unwrap();
|
|
assert_eq!(db.adopt_catalogue(1, &catalogue).await.unwrap(), 0);
|
|
assert_eq!(subs().await, 1);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn every_filter_works_with_and_without_a_search_term() {
|
|
// Regression: the search clause used to be omitted when no term was given, while
|
|
// ?2 was still bound -- rusqlite rejects a parameter the statement never mentions,
|
|
// so plain filtering failed with "Wrong number of parameters passed to query".
|
|
let db = Db::memory().await.unwrap();
|
|
db.exec_for_test(
|
|
"INSERT INTO entries (feed_id, guid, title, description, first_seen, duration) VALUES
|
|
('f','a','Alpha dive','notes one', 100,NULL),
|
|
('f','b','Beta', 'notes two', 200,NULL),
|
|
('f','c','Gamma dive','notes three',300,NULL),
|
|
('f','d','Delta', 'notes four', 400,NULL),
|
|
('f','e','Epsilon', 'notes five', 500,45),
|
|
('f','g','Gimel', 'notes six', 600,NULL),
|
|
('f','h','Heth', 'notes seven',700,900);
|
|
INSERT INTO enclosures (id, feed_id, guid, url, path, state) VALUES
|
|
(1,'f','b','u1','/tmp/b','done');
|
|
-- Read and starred belong to a person now, so say which one.
|
|
INSERT INTO users (id, name, is_admin) VALUES (7,'reader',1);
|
|
INSERT INTO entry_state (user_id, feed_id, guid, read, flagged, position) VALUES
|
|
(7,'f','b',1,0,0),
|
|
(7,'f','c',1,1,0),
|
|
-- Started, length unknown: this is Currently Listening.
|
|
(7,'f','d',0,0,42),
|
|
-- 42 of 45 seconds is past the 90% the player calls finished.
|
|
(7,'f','e',1,0,42),
|
|
-- Barely touched (opened, closed within seconds): not Currently Listening.
|
|
(7,'f','g',0,0,3),
|
|
-- Opened, and so read, but 42 of 900 seconds in: still Currently Listening.
|
|
-- Filtering on read hid exactly these (issue #14).
|
|
(7,'f','h',1,0,42);",
|
|
).await
|
|
.unwrap();
|
|
|
|
for f in [Filter::All, Filter::Unread, Filter::Downloaded, Filter::Flagged, Filter::InProgress] {
|
|
// Both paths must run without erroring, and agree with each other.
|
|
let order = order_sql("published", "desc", false);
|
|
let rows = db.entries_in(7, Some("f"), f, None, 0, 50, &order).await.unwrap();
|
|
let n = db.count_in(7, Some("f"), f, None).await.unwrap();
|
|
assert_eq!(rows.len() as i64, n, "{f:?} count disagrees with the page");
|
|
|
|
let rows = db.entries_in(7, Some("f"), f, Some("dive"), 0, 50, &order).await.unwrap();
|
|
let n = db.count_in(7, Some("f"), f, Some("dive")).await.unwrap();
|
|
assert_eq!(rows.len() as i64, n, "{f:?} with search disagrees");
|
|
}
|
|
|
|
assert_eq!(db.count_in(7, Some("f"), Filter::All, None).await.unwrap(), 7);
|
|
assert_eq!(db.count_in(7, Some("f"), Filter::Unread, None).await.unwrap(), 3);
|
|
assert_eq!(db.count_in(7, Some("f"), Filter::Downloaded, None).await.unwrap(), 1);
|
|
assert_eq!(db.count_in(7, Some("f"), Filter::Flagged, None).await.unwrap(), 1);
|
|
assert_eq!(db.count_in(7, Some("f"), Filter::All, Some("dive")).await.unwrap(), 2);
|
|
assert_eq!(db.count_in(7, Some("f"), Filter::All, Some("NOTES two")).await.unwrap(), 1,
|
|
"search is case-insensitive and covers the description");
|
|
|
|
// Currently Listening: started, not finished, and not just an accidental tap.
|
|
let order = order_sql("published", "desc", false);
|
|
let listening = async || {
|
|
let rows = db.entries_in(7, Some("f"), Filter::InProgress, None, 0, 50, &order).await.unwrap();
|
|
rows.into_iter().map(|e| e.guid).collect::<Vec<_>>()
|
|
};
|
|
assert_eq!(listening().await, ["h", "d"]);
|
|
|
|
// The player's measured length is the one that counts, in place of a missing one or over
|
|
// the feed's: d, 42 of a measured 45 seconds, is finished; e, which its feed calls 45
|
|
// seconds long, is 42 into a 9000-second file and is not. Times left use it too.
|
|
db.set_position(7, "f", "d", 42, Some(45)).await.unwrap();
|
|
db.set_position(7, "f", "e", 42, Some(9000)).await.unwrap();
|
|
assert_eq!(listening().await, ["h", "e"]);
|
|
let rows = db.entries_in(7, Some("f"), Filter::All, None, 0, 50, &order).await.unwrap();
|
|
assert_eq!(rows.iter().find(|r| r.guid == "e").unwrap().duration, Some(9000));
|
|
// A save without one (before the player knows) keeps the length already measured.
|
|
db.set_position(7, "f", "e", 43, None).await.unwrap();
|
|
assert_eq!(listening().await, ["h", "e"]);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn pending_takes_the_latest_episodes_first() {
|
|
// A cap of 3 must mean the three newest, not the three recorded first.
|
|
let db = Db::memory().await.unwrap();
|
|
db.exec_for_test(
|
|
"INSERT INTO entries (feed_id, guid, published, first_seen) VALUES
|
|
('f','old',100,100), ('f','mid',200,200), ('f','new',300,300);
|
|
INSERT INTO enclosures (id, feed_id, guid, url, state) VALUES
|
|
(1,'f','old','u-old','pending'),
|
|
(2,'f','mid','u-mid','pending'),
|
|
(3,'f','new','u-new','pending');",
|
|
).await
|
|
.unwrap();
|
|
let got: Vec<String> = db.pending("f", 2).await.unwrap().into_iter().map(|p| p.url).collect();
|
|
assert_eq!(got, vec!["u-new", "u-mid"], "newest first, oldest left for later");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_restart_requeues_interrupted_downloads() {
|
|
let db = Db::memory().await.unwrap();
|
|
db.exec_for_test(
|
|
"INSERT INTO enclosures (id, feed_id, guid, url, state, path) VALUES
|
|
(1,'f','a','u1','downloading',NULL),
|
|
(2,'f','b','u2','pending',NULL),
|
|
(3,'f','c','u3','downloading','/tmp/already-here'),
|
|
(4,'f','d','u4','done','/tmp/x');",
|
|
).await
|
|
.unwrap();
|
|
assert_eq!(db.requeue_interrupted().await.unwrap(), 1, "only the in-flight, fileless one");
|
|
let state = async |id: i64| db.strings_for_test(&format!("SELECT state FROM enclosures WHERE id = {id}")).await;
|
|
assert_eq!(state(1).await, ["pending"]);
|
|
assert_eq!(state(3).await, ["downloading"], "it has a file; leave it alone");
|
|
assert_eq!(state(4).await, ["done"]);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_show_takes_over_what_its_creator_held() {
|
|
let db = Db::memory().await.unwrap();
|
|
db.exec_for_test(
|
|
"INSERT INTO users (id, name, is_admin) VALUES (1,'ray',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);",
|
|
).await
|
|
.unwrap();
|
|
db.adopt("creator", "show", &[("a", "u1"), ("b", "u2"), ("d", "u4")]).await.unwrap();
|
|
let owner = async |id: i64| db.strings_for_test(&format!("SELECT feed_id FROM enclosures WHERE id = {id}")).await;
|
|
assert_eq!(owner(1).await, ["show"], "a downloaded file moves with its item");
|
|
assert_eq!(owner(2).await, ["show"]);
|
|
assert_eq!(owner(3).await, ["creator"], "this show does not list it");
|
|
assert_eq!(owner(4).await, ["other"], "only the parent's are taken");
|
|
assert_eq!(
|
|
db.strings_for_test("SELECT feed_id FROM entry_state WHERE user_id = 1 AND guid = 'a'").await,
|
|
["show"],
|
|
"what you had read stays read"
|
|
);
|
|
|
|
// Only a filter's verdict can be overturned by a change of settings.
|
|
let skipped = db.skipped_by_filter("show").await.unwrap();
|
|
assert_eq!(skipped.get("u2").map(String::as_str), Some("explicit"));
|
|
assert_eq!(skipped.len(), 1);
|
|
assert!(db.skipped_by_filter("other").await.unwrap().is_empty(), "torrents disabled is not a filter");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_feed_in_a_group_follows_your_settings_on_the_group() {
|
|
let db = Db::memory().await.unwrap();
|
|
db.exec_for_test(
|
|
"INSERT INTO users (id, name, is_admin) VALUES (1,'ray',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);",
|
|
).await
|
|
.unwrap();
|
|
let explicit = async |group| -> Vec<Option<bool>> {
|
|
let mut v: Vec<_> =
|
|
db.subscribers("show", group).await.unwrap().into_iter().map(|s| s.allow_explicit).collect();
|
|
v.sort();
|
|
v
|
|
};
|
|
assert_eq!(explicit(Some("group")).await, [Some(false), Some(true)], "ray inherits; sam's own choice on the show wins");
|
|
assert_eq!(explicit(None).await, [None, Some(false)], "outside a group nothing is inherited");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn error_since_marks_the_start_of_a_run_of_failures_and_clears_on_success() {
|
|
let db = Db::memory().await.unwrap();
|
|
db.set_feed_error("f", "http://x", "HTTP 404").await.unwrap();
|
|
// Backdate it, as if this feed had already been failing a while, so a second
|
|
// failure landing "now" is distinguishable from the first.
|
|
db.exec_for_test("UPDATE feeds SET error_since = error_since - 3600 WHERE id = 'f'").await.unwrap();
|
|
let first = db.feed_summary("f").await.unwrap().error_since.unwrap();
|
|
|
|
// macmanx: failed once, read fine an hour later. A second failure must not push
|
|
// error_since forward -- the UI decides "failing for a day" from the first one.
|
|
db.set_feed_error("f", "http://x", "HTTP 404").await.unwrap();
|
|
assert_eq!(db.feed_summary("f").await.unwrap().error_since, Some(first));
|
|
|
|
db.touch_feed("f", "http://x").await.unwrap();
|
|
let after = db.feed_summary("f").await.unwrap();
|
|
assert_eq!(after.last_error, None);
|
|
assert_eq!(after.error_since, None, "a clean check ends the run of failures");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_pin_survives_saving_the_feeds_settings_and_needs_a_subscription() {
|
|
let db = Db::memory().await.unwrap();
|
|
let me = db.create_user("pat", None, false).await.unwrap();
|
|
assert!(!db.set_pinned(me, "f", true).await.unwrap(), "not subscribed: nothing to pin");
|
|
db.subscribe(me, "f").await.unwrap();
|
|
assert!(db.set_pinned(me, "f", true).await.unwrap());
|
|
// set_subscription writes the rest of the row; it must leave the pin alone.
|
|
db.set_subscription(me, &Sub { feed_id: "f".into(), auto_download: Some(false), ..Default::default() }).await
|
|
.unwrap();
|
|
assert!(db.pinned_feeds(me).await.unwrap().contains("f"));
|
|
db.set_pinned(me, "f", false).await.unwrap();
|
|
assert!(db.pinned_feeds(me).await.unwrap().is_empty());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn enclosure_url_is_the_dedupe_key() {
|
|
let db = Db::memory().await.unwrap();
|
|
let insert = "INSERT INTO enclosures (feed_id, guid, url, state) VALUES ('f', 'g', 'http://x/a.mp3', 'pending')";
|
|
db.exec_for_test(insert).await.unwrap();
|
|
assert!(db.exec_for_test(insert).await.is_err(), "duplicate url must be rejected");
|
|
}
|
|
}
|