rays
4a26c73b82
With max_new_per_check at 0 and no per-subscription limit, the budget is usize::MAX, and pending() bound it `as i64`: -1. SQLite reads LIMIT -1 as no limit; Postgres refuses it, so a feed's downloads failed. The new test fails with "LIMIT must not be negative" on Postgres without the clamp and passes with it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2385 lines
104 KiB
Rust
2385 lines
104 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::{blocklists, catalogue, enclosures, entries, feeds, hidden, sessions, settings, 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),
|
|
schema.create_table_from_entity(catalogue::Entity),
|
|
schema.create_table_from_entity(settings::Entity),
|
|
schema.create_table_from_entity(blocklists::Entity),
|
|
schema.create_table_from_entity(hidden::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(())
|
|
}
|
|
|
|
/// Where the database is: IPX_DATABASE_URL, a postgres:// URL, when it is set; otherwise the
|
|
/// SQLite file in the data directory, as it has always been.
|
|
pub fn location() -> String {
|
|
std::env::var("IPX_DATABASE_URL")
|
|
.ok()
|
|
.filter(|u| !u.trim().is_empty())
|
|
.unwrap_or_else(|| crate::config::data_dir().join("state.db").display().to_string())
|
|
}
|
|
|
|
/// The Postgres connection option that keeps notices from being sent at all; see `connect`.
|
|
const QUIET: &str = "options=-c%20client_min_messages%3Dwarning";
|
|
|
|
/// A URL fit for a log or an error: the password taken out.
|
|
fn redact(url: &str) -> String {
|
|
match (url.find("://"), url.rfind('@')) {
|
|
(Some(s), Some(at)) if at > s => match url[s + 3..at].find(':') {
|
|
Some(c) => format!("{}:***{}", &url[..s + 3 + c], &url[at..]),
|
|
None => url.to_owned(),
|
|
},
|
|
_ => url.to_owned(),
|
|
}
|
|
}
|
|
|
|
fn is_postgres(location: &str) -> bool {
|
|
location.starts_with("postgres://") || location.starts_with("postgresql://")
|
|
}
|
|
|
|
/// 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(location: &str) -> Result<sea_orm::DatabaseConnection> {
|
|
let mut opts = sea_orm::ConnectOptions::new(url_for(location));
|
|
opts.sqlx_logging(false);
|
|
sea_orm::Database::connect(opts)
|
|
.await
|
|
.with_context(|| format!("opening {}", redact(location)))
|
|
}
|
|
|
|
/// The URL `connect` hands sqlx for a location: a SQLite file, created if missing, or a Postgres
|
|
/// URL with notices turned off.
|
|
fn url_for(location: &str) -> String {
|
|
if !is_postgres(location) {
|
|
format!("sqlite://{location}?mode=rwc")
|
|
} else if location.contains("options=") {
|
|
location.to_owned() // someone chose their own; theirs stands
|
|
} else {
|
|
// Warnings and up only. Postgres answers every CREATE ... IF NOT EXISTS on an existing
|
|
// table with a notice, eleven on each open; sqlx logs each, no filter here wants them,
|
|
// and tracing-subscriber's per-layer filters then swallowed the next line ipx logged
|
|
// (0.3.23: "moved the feeds ... into the database" went missing that way).
|
|
let sep = if location.contains('?') { '&' } else { '?' };
|
|
format!("{location}{sep}{QUIET}")
|
|
}
|
|
}
|
|
|
|
|
|
/// 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>,
|
|
/// Words that keep an item from being downloaded for this person: their list for every feed
|
|
/// and their list for this one together. Filled by `subscribers` only; it is kept apart from
|
|
/// the settings above (`Db::set_blocklist`).
|
|
pub blocked: Vec<String>,
|
|
}
|
|
|
|
/// 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,
|
|
blocked: vec![],
|
|
}
|
|
}
|
|
}
|
|
|
|
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 {
|
|
/// Opens the database at `location` (see `location()`): a postgres:// URL, or a SQLite file.
|
|
pub async fn open(location: &str) -> Result<Self> {
|
|
if !is_postgres(location)
|
|
&& let Some(dir) = Path::new(location).parent()
|
|
{
|
|
std::fs::create_dir_all(dir).with_context(|| format!("creating {}", dir.display()))?;
|
|
}
|
|
let orm = connect(location).await?;
|
|
if !is_postgres(location) {
|
|
// 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, empty database for a test, its schema made from the entities as a real one's is.
|
|
/// A SQLite file of its own (two connections to one ":memory:" would be two databases); or,
|
|
/// with IPX_TEST_DATABASE_URL set, a Postgres schema of its own on that database, so the
|
|
/// same tests prove the SQL on both.
|
|
#[cfg(test)]
|
|
pub async fn memory() -> Result<Self> {
|
|
static N: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0);
|
|
let n = N.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
|
|
let pid = std::process::id();
|
|
if let Some(url) = std::env::var("IPX_TEST_DATABASE_URL").ok().filter(|u| !u.is_empty()) {
|
|
let admin = connect(&url).await?;
|
|
// Schemas an earlier run left behind: anything of ours not made by this process.
|
|
let old = admin
|
|
.query_all_raw(Statement::from_string(
|
|
admin.get_database_backend(),
|
|
format!(
|
|
r"SELECT nspname FROM pg_namespace
|
|
WHERE nspname LIKE 'ipxt\_%' AND nspname NOT LIKE 'ipxt\_{pid}\_%'"
|
|
),
|
|
))
|
|
.await?;
|
|
for r in old {
|
|
let name: String = r.try_get_by_index(0)?;
|
|
admin.execute_unprepared(&format!("DROP SCHEMA IF EXISTS {name} CASCADE")).await?;
|
|
}
|
|
let schema = format!("ipxt_{pid}_{n}");
|
|
admin.execute_unprepared(&format!("CREATE SCHEMA {schema}")).await?;
|
|
// Every connection in the pool starts in it, so each test sees only its own tables.
|
|
let sep = if url.contains('?') { '&' } else { '?' };
|
|
let orm =
|
|
connect(&format!("{url}{sep}options=-c%20search_path%3D{schema}%20-c%20client_min_messages%3Dwarning"))
|
|
.await?;
|
|
create_missing(&orm).await?;
|
|
return Ok(Self { orm, tmp: None });
|
|
}
|
|
let path = std::env::temp_dir().join(format!("ipx-test-{pid}-{n}.db"));
|
|
let orm = connect(&path.display().to_string()).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(())
|
|
}
|
|
|
|
#[tracing::instrument(skip_all)]
|
|
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)
|
|
}
|
|
|
|
/// What the feed list shows of every feed, for one person, in five queries whatever the
|
|
/// number of feeds. Asked feed by feed (feed_summary, http_state, blocklist, unread_count)
|
|
/// it was six round trips a feed, about 950 for 160 feeds and 320 ms a page load (#94).
|
|
pub async fn feed_list(&self, user_id: i64) -> Result<std::collections::HashMap<String, FeedListing>> {
|
|
use std::collections::HashMap;
|
|
let counts = |sql: &'static str, args: Vec<sea_orm::Value>| async move {
|
|
self.rows(sql, args)
|
|
.await?
|
|
.iter()
|
|
.map(|r| Ok((r.try_get::<String>("", "feed_id")?, r.try_get::<i64>("", "n")?)))
|
|
.collect::<Result<HashMap<_, _>>>()
|
|
};
|
|
let entries = counts("SELECT feed_id, count(*) AS n FROM entries GROUP BY feed_id", vec![]).await?;
|
|
let downloaded =
|
|
counts("SELECT feed_id, count(*) AS n FROM enclosures WHERE path IS NOT NULL GROUP BY feed_id", vec![])
|
|
.await?;
|
|
let unread = counts(
|
|
"SELECT e.feed_id, count(*) AS n FROM entries e
|
|
JOIN subscriptions sub ON sub.user_id = $1 AND sub.feed_id = e.feed_id
|
|
LEFT JOIN entry_state s
|
|
ON s.user_id = $1 AND s.feed_id = e.feed_id AND s.guid = e.guid
|
|
WHERE NOT coalesce(s.read, false)
|
|
AND NOT EXISTS (SELECT 1 FROM hidden h
|
|
WHERE h.user_id = $1 AND h.feed_id = e.feed_id AND h.guid = e.guid)
|
|
GROUP BY e.feed_id",
|
|
vec![user_id.into()],
|
|
)
|
|
.await?;
|
|
let mut blocked: HashMap<String, Vec<String>> = blocklists::Entity::find()
|
|
.filter(blocklists::Column::UserId.eq(user_id))
|
|
.all(&self.orm)
|
|
.await?
|
|
.into_iter()
|
|
.map(|b| (b.feed_id, keywords(Some(b.words)).unwrap_or_default()))
|
|
.collect();
|
|
Ok(feeds::Entity::find()
|
|
.all(&self.orm)
|
|
.await?
|
|
.into_iter()
|
|
.map(|f| {
|
|
let id = f.id.clone();
|
|
let listing = FeedListing {
|
|
summary: FeedSummary {
|
|
entries: entries.get(&id).copied().unwrap_or(0),
|
|
downloaded: downloaded.get(&id).copied().unwrap_or(0),
|
|
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,
|
|
},
|
|
ttl_mins: f.ttl_mins.map(|t| t.max(0) as u64),
|
|
blocked: blocked.remove(&id).unwrap_or_default(),
|
|
unread: unread.get(&id).copied().unwrap_or(0),
|
|
};
|
|
(id, listing)
|
|
})
|
|
.collect())
|
|
}
|
|
}
|
|
|
|
/// One feed as the feed list shows it; see `Db::feed_list`.
|
|
#[derive(Debug, Default)]
|
|
pub struct FeedListing {
|
|
pub summary: FeedSummary,
|
|
pub ttl_mins: Option<u64>,
|
|
pub blocked: Vec<String>,
|
|
pub unread: i64,
|
|
}
|
|
|
|
|
|
/// 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)]
|
|
#[tracing::instrument(skip_all)]
|
|
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.
|
|
#[tracing::instrument(skip_all)]
|
|
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",
|
|
// Unlimited is usize::MAX, which `as i64` makes -1: SQLite reads LIMIT -1 as no
|
|
// limit, Postgres refuses it, and the feed's downloads failed (#98).
|
|
vec![feed_id.into(), (limit.min(i64::MAX as usize) 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
|
|
-- Someone an item is hidden from is as done with it as someone who read it.
|
|
LEFT JOIN (SELECT feed_id, guid, count(*) AS n
|
|
FROM (SELECT user_id, feed_id, guid FROM entry_state WHERE read
|
|
UNION SELECT user_id, feed_id, guid FROM hidden) done
|
|
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.
|
|
for table in ["entry_state", "hidden"] {
|
|
self.exec(
|
|
&format!(
|
|
"DELETE FROM {table} WHERE NOT EXISTS (
|
|
SELECT 1 FROM entries e
|
|
WHERE e.feed_id = {table}.feed_id AND e.guid = {table}.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)
|
|
AND NOT EXISTS (SELECT 1 FROM hidden h
|
|
WHERE h.user_id = $2 AND h.feed_id = e.feed_id AND h.guid = e.guid)",
|
|
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} AND {}",
|
|
filter.sql(),
|
|
not_hidden(&user)
|
|
)
|
|
}
|
|
|
|
/// Not hidden from this person by their block lists. `e` is entries.
|
|
fn not_hidden(user: &str) -> String {
|
|
format!(
|
|
"NOT EXISTS (SELECT 1 FROM hidden h WHERE h.user_id = {user} AND h.feed_id = e.feed_id AND h.guid = e.guid)"
|
|
)
|
|
}
|
|
|
|
/// 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 items are not lifted above the rest: sorting by the pin column, or the Pinned tab, does
|
|
/// that when it is wanted, and lifting them always was undone (issues #35 and #36).
|
|
pub fn order_sql(col: &str, dir: &str) -> 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)",
|
|
};
|
|
// Where an item with no value goes, said outright: SQLite counts a NULL as the smallest value
|
|
// and Postgres as the largest, so "largest first" on Postgres opened with every item that has
|
|
// no file. NULLS FIRST going up and LAST going down keeps what SQLite did.
|
|
let dir = if dir == "asc" { "ASC NULLS FIRST" } else { "DESC NULLS LAST" };
|
|
// The guid breaks what ties remain: SQLite's rowid did, and Postgres has none.
|
|
format!("{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(())
|
|
}
|
|
|
|
// ---- the configuration: the catalogue and the server settings ----
|
|
|
|
/// The feeds and server settings, once the database holds them; None before, when they are
|
|
/// still config.toml's.
|
|
pub async fn stored_config(
|
|
&self,
|
|
) -> Result<Option<(crate::config::Stored, std::collections::BTreeMap<String, crate::config::Feed>)>> {
|
|
let Some(general) = settings::Entity::find_by_id("general".to_owned()).one(&self.orm).await? else {
|
|
return Ok(None);
|
|
};
|
|
let stored = serde_json::from_str(&general.value).context("reading the stored server settings")?;
|
|
let mut feeds = std::collections::BTreeMap::new();
|
|
for row in catalogue::Entity::find().all(&self.orm).await? {
|
|
let feed = serde_json::from_str(&row.spec).with_context(|| format!("reading feed {}", row.id))?;
|
|
feeds.insert(row.id, feed);
|
|
}
|
|
Ok(Some((stored, feeds)))
|
|
}
|
|
|
|
/// Takes the feeds and server settings from `cfg` (config.toml, as read) into a database that
|
|
/// has none. False when it already had them: another process got there first, and its copy
|
|
/// stands. One transaction, and the settings row goes in first, so two processes starting at
|
|
/// once cannot both import.
|
|
pub async fn import_config(&self, cfg: &crate::config::Config) -> Result<bool> {
|
|
use sea_orm::TransactionTrait;
|
|
let backend = self.orm.get_database_backend();
|
|
let tx = self.orm.begin().await?;
|
|
let claimed = tx
|
|
.execute_raw(Statement::from_sql_and_values(
|
|
backend,
|
|
"INSERT INTO settings (name, value) VALUES ('general', $1) ON CONFLICT DO NOTHING",
|
|
vec![serde_json::to_string(&crate::config::Stored::of(cfg))?.into()],
|
|
))
|
|
.await?
|
|
.rows_affected();
|
|
if claimed == 0 {
|
|
return Ok(false); // dropped, so rolled back
|
|
}
|
|
for (id, feed) in &cfg.feeds {
|
|
catalogue::ActiveModel { id: Set(id.clone()), spec: Set(serde_json::to_string(feed)?) }
|
|
.insert(&tx)
|
|
.await?;
|
|
}
|
|
tx.commit().await?;
|
|
Ok(true)
|
|
}
|
|
|
|
/// Writes the feeds and server settings as they now stand: what config.toml's save did.
|
|
/// The whole catalogue at once, in one transaction, so a feed removed goes too.
|
|
pub async fn store_config(&self, cfg: &crate::config::Config) -> Result<()> {
|
|
use sea_orm::TransactionTrait;
|
|
let backend = self.orm.get_database_backend();
|
|
let tx = self.orm.begin().await?;
|
|
tx.execute_raw(Statement::from_sql_and_values(
|
|
backend,
|
|
"INSERT INTO settings (name, value) VALUES ('general', $1)
|
|
ON CONFLICT (name) DO UPDATE SET value = excluded.value",
|
|
vec![serde_json::to_string(&crate::config::Stored::of(cfg))?.into()],
|
|
))
|
|
.await?;
|
|
catalogue::Entity::delete_many().exec(&tx).await?;
|
|
for (id, feed) in &cfg.feeds {
|
|
catalogue::ActiveModel { id: Set(id.clone()), spec: Set(serde_json::to_string(feed)?) }
|
|
.insert(&tx)
|
|
.await?;
|
|
}
|
|
tx.commit().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.
|
|
#[tracing::instrument(skip_all)]
|
|
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,
|
|
g.words AS global_blocked, b.words AS blocked
|
|
FROM subscriptions c
|
|
LEFT JOIN subscriptions p ON p.user_id = c.user_id AND p.feed_id = $2
|
|
LEFT JOIN blocklists g ON g.user_id = c.user_id AND g.feed_id = ''
|
|
LEFT JOIN blocklists b ON b.user_id = c.user_id AND b.feed_id = c.feed_id
|
|
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")?,
|
|
blocked: [r.try_get("", "global_blocked")?, r.try_get("", "blocked")?]
|
|
.into_iter()
|
|
.flat_map(|w| keywords(w).unwrap_or_default())
|
|
.collect(),
|
|
})
|
|
})
|
|
.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 OR h.guid IS NOT NULL 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
|
|
LEFT JOIN hidden h
|
|
ON h.user_id = s.user_id AND h.feed_id = e.feed_id AND h.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<()> {
|
|
let new = self
|
|
.exec(
|
|
"INSERT INTO subscriptions (user_id, feed_id) VALUES ($1, $2) ON CONFLICT DO NOTHING",
|
|
vec![user_id.into(), feed_id.into()],
|
|
)
|
|
.await?;
|
|
if new > 0 {
|
|
// Their list for every feed applies to this one from the start.
|
|
self.rehide(feed_id).await?;
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// Someone's block list: for one feed, or with `feed_id` empty, for every feed.
|
|
pub async fn blocklist(&self, user_id: i64, feed_id: &str) -> Result<Vec<String>> {
|
|
Ok(blocklists::Entity::find_by_id((user_id, feed_id.to_owned()))
|
|
.one(&self.orm)
|
|
.await?
|
|
.and_then(|b| keywords(Some(b.words)))
|
|
.unwrap_or_default())
|
|
}
|
|
|
|
/// Replaces a block list and works out again what it hides: in that feed, or with
|
|
/// `feed_id` empty, in every feed the person reads.
|
|
pub async fn set_blocklist(&self, user_id: i64, feed_id: &str, words: &[String]) -> Result<()> {
|
|
if words.is_empty() {
|
|
blocklists::Entity::delete_by_id((user_id, feed_id.to_owned())).exec(&self.orm).await?;
|
|
} else {
|
|
self.exec(
|
|
"INSERT INTO blocklists (user_id, feed_id, words) VALUES ($1, $2, $3)
|
|
ON CONFLICT (user_id, feed_id) DO UPDATE SET words = excluded.words",
|
|
vec![user_id.into(), feed_id.into(), serde_json::to_string(words)?.into()],
|
|
)
|
|
.await?;
|
|
}
|
|
if feed_id.is_empty() {
|
|
for sub in self.subscriptions_for(user_id).await? {
|
|
self.rehide(&sub.feed_id).await?;
|
|
}
|
|
} else {
|
|
self.rehide(feed_id).await?;
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// Works out again which of a feed's items each subscriber's block lists hide from them.
|
|
///
|
|
/// ponytail: every item of the feed for every subscriber, each time the feed is scanned with
|
|
/// something new or a list changes. Fine at hundreds of items a feed; look only at new items
|
|
/// on a scan if a feed with thousands makes scans slow.
|
|
#[tracing::instrument(skip_all)]
|
|
pub async fn rehide(&self, feed_id: &str) -> Result<()> {
|
|
let lists: Vec<(i64, Vec<String>)> = self
|
|
.rows(
|
|
"SELECT s.user_id, g.words AS global, b.words AS feed
|
|
FROM subscriptions s
|
|
LEFT JOIN blocklists g ON g.user_id = s.user_id AND g.feed_id = ''
|
|
LEFT JOIN blocklists b ON b.user_id = s.user_id AND b.feed_id = s.feed_id
|
|
WHERE s.feed_id = $1",
|
|
vec![feed_id.into()],
|
|
)
|
|
.await?
|
|
.iter()
|
|
.map(|r| {
|
|
let words = [r.try_get("", "global")?, r.try_get("", "feed")?]
|
|
.into_iter()
|
|
.flat_map(|w: Option<String>| keywords(w).unwrap_or_default())
|
|
.collect();
|
|
Ok((r.try_get("", "user_id")?, words))
|
|
})
|
|
.collect::<Result<_>>()?;
|
|
hidden::Entity::delete_many().filter(hidden::Column::FeedId.eq(feed_id)).exec(&self.orm).await?;
|
|
let lists: Vec<_> = lists.into_iter().filter(|(_, w)| !w.is_empty()).collect();
|
|
if lists.is_empty() {
|
|
return Ok(());
|
|
}
|
|
let items = entries::Entity::find().filter(entries::Column::FeedId.eq(feed_id)).all(&self.orm).await?;
|
|
let rows: Vec<hidden::ActiveModel> = items
|
|
.iter()
|
|
.flat_map(|e| {
|
|
let hay = [e.title.as_deref().unwrap_or(""), e.description.as_deref().unwrap_or("")];
|
|
lists
|
|
.iter()
|
|
.filter(move |(_, words)| crate::download::blocked(words, &hay))
|
|
.map(|(user, _)| hidden::ActiveModel {
|
|
user_id: Set(*user),
|
|
feed_id: Set(feed_id.to_owned()),
|
|
guid: Set(e.guid.clone()),
|
|
})
|
|
})
|
|
.collect();
|
|
if !rows.is_empty() {
|
|
hidden::Entity::insert_many(rows).exec_without_returning(&self.orm).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.
|
|
/// Whether a feed or an entry names `url` as its artwork: the only addresses /api/art will
|
|
/// fetch, so the web server cannot be pointed at anything else.
|
|
// ponytail: entries.image is unindexed, a scan per http:// image; index it if that shows up.
|
|
pub async fn names_image(&self, url: &str) -> Result<bool> {
|
|
Ok(feeds::Entity::find().filter(feeds::Column::Image.eq(url)).count(&self.orm).await? > 0
|
|
|| entries::Entity::find().filter(entries::Column::Image.eq(url)).count(&self.orm).await? > 0)
|
|
}
|
|
|
|
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 when it was kept on the account, and light, dark or auto; None
|
|
/// for either if they never did. Only read now, to give a browser with no theme cookie its
|
|
/// first one (issue #69).
|
|
// ponytail: users.theme and theme_mode are never written any more; drop them once every
|
|
// browser in use has its own cookie.
|
|
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())
|
|
}
|
|
|
|
/// 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(())
|
|
}
|
|
|
|
/// 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.
|
|
#[tracing::instrument(skip_all)]
|
|
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.
|
|
/// The guids of a feed's stored items and the URLs of its files: a scan inserts only what is
|
|
/// not among them. Inserting every item it read, stored or not, cost a round trip each, about
|
|
/// 10 ms: 13 s a scan for Clarkesworld's 1200 items, and 117 s of a full 315 s scan (#96).
|
|
#[tracing::instrument(skip_all)]
|
|
pub async fn stored_items(
|
|
&self,
|
|
feed_id: &str,
|
|
) -> Result<(std::collections::HashSet<String>, std::collections::HashSet<String>)> {
|
|
let col = |sql: &'static str, name: &'static str| async move {
|
|
self.rows(sql, vec![feed_id.into()])
|
|
.await?
|
|
.iter()
|
|
.map(|r| Ok(r.try_get::<String>("", name)?))
|
|
.collect::<Result<std::collections::HashSet<_>>>()
|
|
};
|
|
Ok((
|
|
col("SELECT guid FROM entries WHERE feed_id = $1", "guid").await?,
|
|
col("SELECT url FROM enclosures WHERE feed_id = $1", "url").await?,
|
|
))
|
|
}
|
|
|
|
#[tracing::instrument(skip_all)]
|
|
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 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',true);
|
|
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)).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"]);
|
|
// An item with no file has no size or type: last going down, first going up, on both
|
|
// databases (they disagree about NULL unless told).
|
|
db.exec_for_test("INSERT INTO entries (feed_id, guid, title, first_seen) VALUES ('f','n','no file',50);")
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(order("size", "desc").await.last().map(String::as_str), Some("n"));
|
|
assert_eq!(order("type", "asc").await.first().map(String::as_str), Some("n"));
|
|
db.exec_for_test("DELETE FROM entries WHERE guid = 'n';").await.unwrap();
|
|
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");
|
|
// A pin does not lift an item above the others in any other sort (#36).
|
|
assert_eq!(order("published", "desc").await, ["c", "b", "a"]);
|
|
// 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").contains("x'"));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn the_feed_list_in_one_go_matches_asking_feed_by_feed() {
|
|
let db = Db::memory().await.unwrap();
|
|
db.exec_for_test(
|
|
"INSERT INTO users (id, name, is_admin) VALUES (1,'ray',true),(2,'sam',false);
|
|
INSERT INTO subscriptions (user_id, feed_id) VALUES (1,'f'),(1,'g'),(2,'f');
|
|
INSERT INTO feeds (id, url, title, ttl_mins) VALUES ('f','u','F',30),('g','v','G',null);
|
|
INSERT INTO entries (feed_id, guid, first_seen) VALUES ('f','a',0),('f','b',0),('f','c',0),('g','d',0);
|
|
INSERT INTO entry_state (user_id, feed_id, guid, read) VALUES (1,'f','a',true),(2,'f','b',true);
|
|
INSERT INTO hidden (user_id, feed_id, guid) VALUES (1,'f','c');
|
|
INSERT INTO enclosures (id, feed_id, guid, url, path, state) VALUES
|
|
(1,'f','a','u1','/tmp/a','done'),(2,'f','b','u2',null,'pending');
|
|
INSERT INTO blocklists (user_id, feed_id, words) VALUES (1,'g','[\"spoiler\"]');",
|
|
).await
|
|
.unwrap();
|
|
let list = db.feed_list(1).await.unwrap();
|
|
for id in ["f", "g"] {
|
|
let one = &list[id];
|
|
let s = db.feed_summary(id).await.unwrap();
|
|
assert_eq!((one.summary.entries, one.summary.downloaded), (s.entries, s.downloaded), "{id}");
|
|
assert_eq!(one.unread, db.unread_count(1, id).await.unwrap(), "{id}");
|
|
assert_eq!(one.blocked, db.blocklist(1, id).await.unwrap(), "{id}");
|
|
assert_eq!(one.ttl_mins, db.http_state(id).await.unwrap().ttl_mins, "{id}");
|
|
}
|
|
assert_eq!((list["f"].unread, list["g"].unread), (1, 1)); // a read, c hidden; sam's read is sam's
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn an_unlimited_queue_takes_everything_waiting() {
|
|
let db = Db::memory().await.unwrap();
|
|
db.exec_for_test(
|
|
"INSERT INTO entries (feed_id, guid, first_seen) VALUES ('f','a',0),('f','b',0);
|
|
INSERT INTO enclosures (id, feed_id, guid, url, state) VALUES (1,'f','a','u1','pending'),(2,'f','b','u2','pending');",
|
|
).await
|
|
.unwrap();
|
|
assert_eq!(db.pending("f", usize::MAX).await.unwrap().len(), 2);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_scan_knows_a_feeds_stored_items_and_files() {
|
|
let db = Db::memory().await.unwrap();
|
|
db.exec_for_test(
|
|
"INSERT INTO entries (feed_id, guid, first_seen) VALUES ('f','a',0),('f','b',0),('g','c',0);
|
|
INSERT INTO enclosures (id, feed_id, guid, url, state) VALUES (1,'f','a','u1','pending'),(2,'g','c','u2','pending');",
|
|
).await
|
|
.unwrap();
|
|
let (items, files) = db.stored_items("f").await.unwrap();
|
|
assert_eq!(items, ["a", "b"].map(String::from).into());
|
|
assert_eq!(files, ["u1"].map(String::from).into()); // u2 is g's: left to the insert to find
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn only_artwork_a_feed_names_is_fetched_for_the_page() {
|
|
let db = Db::memory().await.unwrap();
|
|
db.exec_for_test(
|
|
"INSERT INTO feeds (id, url, image) VALUES ('f','u','http://cdn.example/show.jpg');
|
|
INSERT INTO entries (feed_id, guid, first_seen, image) VALUES ('f','a',0,'http://cdn.example/ep.jpg');",
|
|
).await
|
|
.unwrap();
|
|
assert!(db.names_image("http://cdn.example/show.jpg").await.unwrap());
|
|
assert!(db.names_image("http://cdn.example/ep.jpg").await.unwrap());
|
|
assert!(!db.names_image("http://192.168.1.1/admin").await.unwrap());
|
|
}
|
|
|
|
#[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',true),(2,'sam',false),(3,'kit',false);
|
|
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',true),(2,'sam',false);
|
|
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");
|
|
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',true);").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',true);
|
|
INSERT INTO entry_state (user_id, feed_id, guid, read, flagged, position) VALUES
|
|
(7,'f','b',true,false,0),
|
|
(7,'f','c',true,true,0),
|
|
-- Started, length unknown: this is Currently Listening.
|
|
(7,'f','d',false,false,42),
|
|
-- 42 of 45 seconds is past the 90% the player calls finished.
|
|
(7,'f','e',true,false,42),
|
|
-- Barely touched (opened, closed within seconds): not Currently Listening.
|
|
(7,'f','g',false,false,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',true,false,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");
|
|
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");
|
|
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 block_lists_hide_items_from_whoever_keeps_them() {
|
|
let db = Db::memory().await.unwrap();
|
|
db.exec_for_test(
|
|
"INSERT INTO users (id, name, is_admin) VALUES (1,'a',true), (2,'b',false);
|
|
INSERT INTO subscriptions (user_id, feed_id) VALUES (1,'f'), (2,'f');
|
|
INSERT INTO entries (feed_id, guid, title, description, first_seen) VALUES
|
|
('f','x','Election night','',1),
|
|
('f','y','Baking bread','<p>No politics here, honest</p>',2),
|
|
('f','z','Gardening','',3);",
|
|
)
|
|
.await
|
|
.unwrap();
|
|
let shown = async |user| db.count_in(user, Some("f"), Filter::All, None).await.unwrap();
|
|
let words = |w: &[&str]| w.iter().map(|s| s.to_string()).collect::<Vec<_>>();
|
|
|
|
db.set_blocklist(1, "", &words(&["election"])).await.unwrap();
|
|
db.set_blocklist(1, "f", &words(&["politics"])).await.unwrap();
|
|
assert_eq!(shown(1).await, 1, "a title and a body each hide an item, both lists apply");
|
|
assert_eq!(db.unread_count(1, "f").await.unwrap(), 1, "hidden is not unread");
|
|
assert_eq!(shown(2).await, 3, "someone else's list hides nothing from you");
|
|
|
|
db.set_blocklist(1, "", &[]).await.unwrap();
|
|
assert_eq!(shown(1).await, 2, "emptying a list brings its items back");
|
|
assert!(db.blocklist(1, "").await.unwrap().is_empty());
|
|
|
|
// Someone subscribing starts with their list applied, and the scanner sees it.
|
|
db.set_blocklist(2, "", &words(&["gardening"])).await.unwrap();
|
|
db.exec_for_test("INSERT INTO entries (feed_id, guid, title, first_seen) VALUES ('g','w','Gardening',4)")
|
|
.await
|
|
.unwrap();
|
|
db.subscribe(2, "g").await.unwrap();
|
|
assert_eq!(db.count_in(2, Some("g"), Filter::All, None).await.unwrap(), 0);
|
|
assert_eq!(db.subscribers("g", None).await.unwrap()[0].blocked, ["gardening"]);
|
|
}
|
|
|
|
#[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',true);
|
|
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',true);",
|
|
).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',true),(2,'sam',false);
|
|
INSERT INTO subscriptions (user_id, feed_id, allow_explicit) VALUES
|
|
(1,'group',true),(1,'show',NULL),(2,'group',true),(2,'show',false);",
|
|
).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());
|
|
}
|
|
|
|
#[test]
|
|
fn postgres_is_asked_for_no_notices_and_a_password_never_reaches_a_log() {
|
|
assert_eq!(url_for("postgres://u:p@h/d"), format!("postgres://u:p@h/d?{QUIET}"));
|
|
assert_eq!(url_for("postgres://u:p@h/d?sslmode=disable"), format!("postgres://u:p@h/d?sslmode=disable&{QUIET}"));
|
|
assert_eq!(url_for("postgres://u:p@h/d?options=-c%20x%3Dy"), "postgres://u:p@h/d?options=-c%20x%3Dy", "theirs stands");
|
|
assert_eq!(url_for("/data/state.db"), "sqlite:///data/state.db?mode=rwc");
|
|
assert_eq!(redact("postgres://ipodderx:s3cret@h:5433/ipodderx"), "postgres://ipodderx:***@h:5433/ipodderx");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn the_configuration_goes_in_once_and_comes_back_as_it_went() {
|
|
let db = Db::memory().await.unwrap();
|
|
assert!(db.stored_config().await.unwrap().is_none(), "nothing until it is taken in");
|
|
let mut cfg: crate::config::Config = toml::from_str(
|
|
r#"
|
|
[general]
|
|
schedule = "every 2h"
|
|
max_total_gb = 1.5
|
|
[feeds.a]
|
|
url = "http://x/a.xml"
|
|
keywords = ["one", "two"]
|
|
[feeds.b]
|
|
url = "http://x/b.xml"
|
|
password = "secret"
|
|
"#,
|
|
)
|
|
.unwrap();
|
|
assert!(db.import_config(&cfg).await.unwrap());
|
|
assert!(!db.import_config(&cfg).await.unwrap(), "a second import finds the first and stands back");
|
|
let (stored, feeds) = db.stored_config().await.unwrap().unwrap();
|
|
assert_eq!(stored, crate::config::Stored::of(&cfg));
|
|
assert_eq!(feeds.keys().collect::<Vec<_>>(), ["a", "b"]);
|
|
assert_eq!(feeds["a"].keywords, ["one", "two"]);
|
|
assert_eq!(feeds["b"].password.as_deref(), Some("secret"));
|
|
|
|
// Storing is the whole catalogue: a feed taken out goes, one put in arrives.
|
|
cfg.feeds.remove("a");
|
|
cfg.feeds.insert("c".into(), cfg.feeds["b"].clone());
|
|
cfg.general.schedule = "every 30m".into();
|
|
db.store_config(&cfg).await.unwrap();
|
|
let (stored, feeds) = db.stored_config().await.unwrap().unwrap();
|
|
assert_eq!(stored.schedule, "every 30m");
|
|
assert_eq!(feeds.keys().collect::<Vec<_>>(), ["b", "c"]);
|
|
}
|
|
|
|
#[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");
|
|
}
|
|
}
|