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