ipodderx-rs/src/db.rs

//! 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::{catalogue, enclosures, entries, feeds, 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),
    ] {
        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())
}

/// 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>,
}

/// 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}%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(())
    }

    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 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(())
    }

    // ---- 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?),
            ("catalogue", copy_table::<catalogue::Entity>(&from.orm, &tx).await?),
            ("settings", copy_table::<settings::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)).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 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 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");
    }
}