//! SQLite state. Replaces the per-feed .ipxd plists, history.dat and qmcache.dat.

use anyhow::{Context, Result};
use rusqlite::{Connection, OptionalExtension};
use std::path::Path;
use std::sync::Mutex;

/// ponytail: one global connection mutex. Writes here are tiny and rare; move to a
/// spawn_blocking pool if a large feed count ever makes it contend.
pub struct Db {
    conn: Mutex<Connection>,
}

const SCHEMA: &str = "
CREATE TABLE IF NOT EXISTS feeds (
    id            TEXT PRIMARY KEY,
    url           TEXT NOT NULL,
    title         TEXT,
    image         TEXT,
    etag          TEXT,
    last_modified TEXT,
    last_checked  INTEGER,
    ttl_mins      INTEGER,
    last_error    TEXT,
    -- Came from a subscribed OPML that no longer lists it, but has downloads, so kept.
    orphaned      INTEGER NOT NULL DEFAULT 0,
    -- The OPML subscription this feed came from.
    group_id      TEXT,
    -- 1 = derived from an OPML and not written to config.toml. Writing 80-odd generated
    -- entries into a hand-edited file made it unreadable; the OPML is the source of
    -- truth, so they are re-derived instead. Customising one promotes it to config.
    managed       INTEGER NOT NULL DEFAULT 0
);

CREATE TABLE IF NOT EXISTS entries (
    feed_id     TEXT    NOT NULL,
    guid        TEXT    NOT NULL,
    title       TEXT,
    link        TEXT,
    published   INTEGER,
    description TEXT,
    first_seen  INTEGER NOT NULL,
    read        INTEGER NOT NULL DEFAULT 0,
    flagged     INTEGER NOT NULL DEFAULT 0,
    image       TEXT,
    duration    INTEGER,
    episode     INTEGER,
    season      INTEGER,
    -- Seconds into the audio, so playback resumes where it was left.
    position    INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (feed_id, guid)
);

-- url is UNIQUE: this is the dedupe key, and it subsumes the old history.dat pickle.
-- A reaped file keeps its row with path = NULL and state = 'reaped', so a purged
-- episode is never fetched a second time.
CREATE TABLE IF NOT EXISTS enclosures (
    id            INTEGER PRIMARY KEY,
    feed_id       TEXT    NOT NULL,
    guid          TEXT    NOT NULL,
    url           TEXT    NOT NULL UNIQUE,
    mime          TEXT,
    length        INTEGER,
    path          TEXT,
    state         TEXT    NOT NULL,
    bytes_done    INTEGER NOT NULL DEFAULT 0,
    downloaded_at INTEGER,
    last_error    TEXT
);

CREATE INDEX IF NOT EXISTS enclosures_entry ON enclosures (feed_id, guid);
";

/// 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,
    pub orphaned: bool,
}

/// Adds columns that later versions introduced. CREATE TABLE IF NOT EXISTS does nothing to
/// a table that already exists, so an installed database needs them added explicitly.
fn migrate(conn: &Connection) -> Result<()> {
    let wanted: &[(&str, &str, &str)] = &[
        ("feeds", "image", "TEXT"),
        ("feeds", "orphaned", "INTEGER NOT NULL DEFAULT 0"),
        ("feeds", "group_id", "TEXT"),
        ("feeds", "managed", "INTEGER NOT NULL DEFAULT 0"),
        ("entries", "image", "TEXT"),
        ("entries", "duration", "INTEGER"),
        ("entries", "episode", "INTEGER"),
        ("entries", "season", "INTEGER"),
        ("entries", "position", "INTEGER NOT NULL DEFAULT 0"),
    ];
    for (table, column, ty) in wanted {
        let mut stmt = conn.prepare(&format!("PRAGMA table_info({table})"))?;
        let existing: Vec<String> = stmt
            .query_map([], |r| r.get::<_, String>(1))?
            .collect::<rusqlite::Result<Vec<_>>>()?;
        if !existing.iter().any(|c| c == column) {
            tracing::info!(table, column, "adding column");
            conn.execute_batch(&format!("ALTER TABLE {table} ADD COLUMN {column} {ty}"))?;
        }
    }
    Ok(())
}

/// What `ipx list` shows next to each configured feed.
#[derive(Debug, Default)]
pub struct FeedSummary {
    pub title: Option<String>,
    pub image: Option<String>,
    /// Came from a subscribed OPML that no longer lists it, but it has downloads, so it
    /// was kept rather than removed.
    pub orphaned: bool,
    pub last_checked: Option<i64>,
    pub last_error: Option<String>,
    pub entries: i64,
    pub downloaded: i64,
}

impl Db {
    pub fn open(path: &Path) -> Result<Self> {
        if let Some(dir) = path.parent() {
            std::fs::create_dir_all(dir)
                .with_context(|| format!("creating {}", dir.display()))?;
        }
        let conn = Connection::open(path)
            .with_context(|| format!("opening {}", path.display()))?;
        conn.pragma_update(None, "journal_mode", "WAL")?;
        conn.pragma_update(None, "foreign_keys", "ON")?;
        conn.pragma_update(None, "busy_timeout", 5000)?;
        conn.execute_batch(SCHEMA).context("creating schema")?;
        migrate(&conn).context("migrating schema")?;
        Ok(Self { conn: Mutex::new(conn) })
    }

    /// In-memory database, for tests.
    #[cfg(test)]
    pub fn memory() -> Result<Self> {
        let conn = Connection::open_in_memory()?;
        conn.execute_batch(SCHEMA)?;
        // Same path as a real open, so a column added only in migrate() cannot pass the
        // tests while being missing in production (or the reverse).
        migrate(&conn)?;
        Ok(Self { conn: Mutex::new(conn) })
    }

    #[cfg(test)]
    pub fn exec_for_test(&self, sql: &str) -> Result<()> {
        self.conn.lock().unwrap().execute_batch(sql)?;
        Ok(())
    }

    pub fn feed_summary(&self, feed_id: &str) -> Result<FeedSummary> {
        let conn = self.conn.lock().unwrap();
        let mut sum: FeedSummary = conn
            .query_row(
                "SELECT title, image, last_checked, last_error, coalesce(orphaned, 0)
                 FROM feeds WHERE id = ?1",
                [feed_id],
                |r| {
                    Ok(FeedSummary {
                        title: r.get(0)?,
                        image: r.get(1)?,
                        last_checked: r.get(2)?,
                        last_error: r.get(3)?,
                        orphaned: r.get::<_, i64>(4)? != 0,
                        ..Default::default()
                    })
                },
            )
            .optional()?
            .unwrap_or_default();

        sum.entries = conn.query_row(
            "SELECT count(*) FROM entries WHERE feed_id = ?1",
            [feed_id],
            |r| r.get(0),
        )?;
        sum.downloaded = conn.query_row(
            "SELECT count(*) FROM enclosures WHERE feed_id = ?1 AND path IS NOT NULL",
            [feed_id],
            |r| r.get(0),
        )?;
        Ok(sum)
    }
}


/// Everything `fetch` needs to decide whether to poll a feed, and how.
#[derive(Debug, Default)]
pub struct HttpState {
    pub etag: Option<String>,
    pub last_modified: Option<String>,
    pub last_checked: Option<i64>,
    pub ttl_mins: Option<u64>,
}

impl Db {
    pub fn http_state(&self, feed_id: &str) -> Result<HttpState> {
        let conn = self.conn.lock().unwrap();
        Ok(conn
            .query_row(
                "SELECT etag, last_modified, last_checked, ttl_mins FROM feeds WHERE id = ?1",
                [feed_id],
                |r| {
                    Ok(HttpState {
                        etag: r.get(0)?,
                        last_modified: r.get(1)?,
                        last_checked: r.get(2)?,
                        ttl_mins: r.get::<_, Option<i64>>(3)?.map(|t| t.max(0) as u64),
                    })
                },
            )
            .optional()?
            .unwrap_or_default())
    }

    /// Upsert after a successful poll. Clears any previous error.
    pub fn record_feed(
        &self,
        feed_id: &str,
        url: &str,
        title: Option<&str>,
        etag: Option<&str>,
        last_modified: Option<&str>,
        ttl_mins: Option<u64>,
        image: Option<&str>,
    ) -> Result<()> {
        let conn = self.conn.lock().unwrap();
        conn.execute(
            "INSERT INTO feeds (id, url, title, etag, last_modified, last_checked, ttl_mins, last_error, image)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, NULL, ?8)
             ON CONFLICT(id) DO UPDATE SET
                url = excluded.url,
                title = coalesce(excluded.title, feeds.title),
                etag = excluded.etag,
                last_modified = excluded.last_modified,
                last_checked = excluded.last_checked,
                ttl_mins = excluded.ttl_mins,
                image = coalesce(excluded.image, feeds.image),
                last_error = NULL",
            rusqlite::params![feed_id, url, title, etag, last_modified, now(), ttl_mins.map(|t| t as i64), image],
        )?;
        Ok(())
    }

    /// 304, or any other poll that produced no new data: only the clock moves.
    pub fn touch_feed(&self, feed_id: &str, url: &str) -> Result<()> {
        let conn = self.conn.lock().unwrap();
        conn.execute(
            "INSERT INTO feeds (id, url, last_checked) VALUES (?1, ?2, ?3)
             ON CONFLICT(id) DO UPDATE SET last_checked = excluded.last_checked, last_error = NULL",
            rusqlite::params![feed_id, url, now()],
        )?;
        Ok(())
    }

    /// Forgets the cached ETag/Last-Modified. Those validators belong to the old URL, so
    /// keeping them across a URL change could produce a bogus 304 against the new one.
    pub fn clear_validators(&self, feed_id: &str) -> Result<()> {
        let conn = self.conn.lock().unwrap();
        conn.execute(
            "UPDATE feeds SET etag = NULL, last_modified = NULL, last_checked = NULL WHERE id = ?1",
            [feed_id],
        )?;
        Ok(())
    }

    pub fn set_feed_error(&self, feed_id: &str, url: &str, msg: &str) -> Result<()> {
        let conn = self.conn.lock().unwrap();
        conn.execute(
            "INSERT INTO feeds (id, url, last_checked, last_error) VALUES (?1, ?2, ?3, ?4)
             ON CONFLICT(id) DO UPDATE SET last_checked = excluded.last_checked, last_error = excluded.last_error",
            rusqlite::params![feed_id, url, now(), msg],
        )?;
        Ok(())
    }

    /// Returns true when this entry had not been seen before.
    ///
    /// A changed description or title flips `read` back to 0, which is what the original's
    /// textDiff dance was ultimately for -- minus the diff markup, which the UI can do.
    pub fn record_entry(&self, feed_id: &str, e: &crate::feed::Entry) -> Result<bool> {
        let conn = self.conn.lock().unwrap();
        let inserted = conn.execute(
            "INSERT OR IGNORE INTO entries
                (feed_id, guid, title, link, published, description, first_seen, read, flagged,
                 image, duration, episode, season)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, 0, 0, ?8, ?9, ?10, ?11)",
            rusqlite::params![
                feed_id, e.guid, e.title, e.link, e.published, e.description, now(),
                e.image, e.duration, e.episode, e.season
            ],
        )?;
        if inserted == 0 {
            // The SET expressions see the pre-update row, so this compares old vs new.
            conn.execute(
                "UPDATE entries SET
                    title = coalesce(?3, title),
                    description = coalesce(?4, description),
                    image = coalesce(?5, image),
                    duration = coalesce(?6, duration),
                    episode = coalesce(?7, episode),
                    season = coalesce(?8, season),
                    read = CASE WHEN description IS NOT ?4 OR title IS NOT ?3 THEN 0 ELSE read END
                 WHERE feed_id = ?1 AND guid = ?2",
                rusqlite::params![
                    feed_id, e.guid, e.title, e.description,
                    e.image, e.duration, e.episode, e.season
                ],
            )?;
        }
        Ok(inserted == 1)
    }

    /// Returns true when this enclosure URL is new. False means we have downloaded it
    /// before, or deliberately reaped it -- either way it is not fetched again.

    pub fn mark_downloaded(&self, url: &str, path: &std::path::Path, bytes: u64) -> Result<()> {
        let conn = self.conn.lock().unwrap();
        conn.execute(
            "UPDATE enclosures SET state = 'done', path = ?2, bytes_done = ?3,
                downloaded_at = ?4, last_error = NULL WHERE url = ?1",
            rusqlite::params![url, path.to_string_lossy(), bytes as i64, now()],
        )?;
        Ok(())
    }

    /// The row stays -- a failed URL is still a URL we have seen. `state` says why it has
    /// no file, and a retry is an explicit act rather than something a rescan does silently.
    pub fn mark_enclosure(&self, url: &str, state: &str, error: Option<&str>) -> Result<()> {
        let conn = self.conn.lock().unwrap();
        conn.execute(
            "UPDATE enclosures SET state = ?2, last_error = ?3 WHERE url = ?1",
            rusqlite::params![url, state, error],
        )?;
        Ok(())
    }

    pub fn record_enclosure(
        &self,
        feed_id: &str,
        guid: &str,
        enc: &crate::feed::Enclosure,
    ) -> Result<bool> {
        let conn = self.conn.lock().unwrap();
        let inserted = conn.execute(
            "INSERT OR IGNORE INTO enclosures (feed_id, guid, url, mime, length, state)
             VALUES (?1, ?2, ?3, ?4, ?5, 'pending')",
            rusqlite::params![feed_id, guid, enc.url, enc.mime, enc.length],
        )?;
        Ok(inserted == 1)
    }
}


/// An enclosure waiting to be downloaded.
#[derive(Debug)]
pub struct Pending {
    pub id: i64,
    pub url: String,
    pub mime: Option<String>,
}

impl Db {
    /// The download queue is the table, not the parse result: an enclosure held back by
    /// `max_new_per_check` is simply picked up by the next scan, in feed order.
    pub fn pending(&self, feed_id: &str, limit: usize) -> Result<Vec<Pending>> {
        let conn = self.conn.lock().unwrap();
        let mut stmt = conn.prepare(
            // Newest first: a cap of 3 should mean the three latest episodes, not the
            // three that happen to have been recorded first.
            "SELECT x.id, x.url, x.mime FROM enclosures x
             JOIN entries e ON e.feed_id = x.feed_id AND e.guid = x.guid
             WHERE x.feed_id = ?1 AND x.state = 'pending'
             ORDER BY coalesce(e.published, e.first_seen) DESC, x.id DESC
             LIMIT ?2",
        )?;
        let rows = stmt
            .query_map(rusqlite::params![feed_id, limit as i64], |r| {
                Ok(Pending { id: r.get(0)?, url: r.get(1)?, mime: r.get(2)? })
            })?
            .collect::<rusqlite::Result<Vec<_>>>()?;
        Ok(rows)
    }
}


/// A downloaded file the reaper may consider.
#[derive(Debug, Clone, PartialEq)]
pub struct Candidate {
    pub id: i64,
    pub url: String,
    pub path: String,
    pub bytes: i64,
    /// When it landed; the reaper works oldest-first.
    pub age_key: i64,
    pub read: bool,
}

impl Db {
    /// Files on disk, flagged ones excluded, read before unread and oldest first within
    /// each group.
    ///
    /// 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). Requiring
    /// `read = 1` outright would be just as dead here, since nothing marks episodes read
    /// until a UI exists -- so `flagged` is the keep-forever marker, and `read` only decides
    /// what goes first.
    pub fn reap_candidates(&self) -> Result<Vec<Candidate>> {
        let conn = self.conn.lock().unwrap();
        let mut stmt = conn.prepare(
            "SELECT e.id, e.url, e.path, e.bytes_done,
                    coalesce(e.downloaded_at, 0), coalesce(n.read, 0), coalesce(n.flagged, 0)
             FROM enclosures e
             LEFT JOIN entries n ON n.feed_id = e.feed_id AND n.guid = e.guid
             WHERE e.path IS NOT NULL AND coalesce(n.flagged, 0) = 0
             ORDER BY coalesce(n.read, 0) DESC, coalesce(e.downloaded_at, 0) ASC, e.id ASC",
        )?;
        let rows = stmt
            .query_map([], |r| {
                Ok(Candidate {
                    id: r.get(0)?,
                    url: r.get(1)?,
                    path: r.get(2)?,
                    bytes: r.get(3)?,
                    age_key: r.get(4)?,
                    read: r.get::<_, i64>(5)? != 0,
                })
            })?
            .collect::<rusqlite::Result<Vec<_>>>()?;
        Ok(rows)
    }

    /// The row survives the file: that is what stops a reaped episode being re-downloaded.
    pub fn mark_reaped(&self, id: i64) -> Result<()> {
        let conn = self.conn.lock().unwrap();
        conn.execute(
            "UPDATE enclosures SET state = 'reaped', path = NULL WHERE id = ?1",
            [id],
        )?;
        Ok(())
    }

    /// Rows claiming a file that is no longer there (someone deleted it by hand).
    pub fn missing_files(&self) -> Result<Vec<(i64, String)>> {
        let conn = self.conn.lock().unwrap();
        let mut stmt =
            conn.prepare("SELECT id, path FROM enclosures WHERE path IS NOT NULL")?;
        let rows = stmt
            .query_map([], |r| Ok((r.get(0)?, r.get::<_, String>(1)?)))?
            .collect::<rusqlite::Result<Vec<_>>>()?;
        Ok(rows
            .into_iter()
            .filter(|(_, p)| !std::path::Path::new(p).exists())
            .collect())
    }

    /// Old entries that never had a file, or no longer have one. Enclosure rows stay --
    /// they are the dedupe history.
    pub fn prune_entries(&self, older_than: i64) -> Result<usize> {
        let conn = self.conn.lock().unwrap();
        let n = conn.execute(
            "DELETE FROM entries WHERE flagged = 0
               AND 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)",
            [older_than],
        )?;
        Ok(n)
    }
}

impl Db {
    pub fn unread_count(&self, feed_id: &str) -> Result<i64> {
        let conn = self.conn.lock().unwrap();
        Ok(conn.query_row(
            "SELECT count(*) FROM entries WHERE feed_id = ?1 AND read = 0",
            [feed_id],
            |r| r.get(0),
        )?)
    }

    /// (pending, downloaded) across all feeds, for the status command.
    pub fn counts(&self) -> Result<(i64, i64)> {
        let conn = self.conn.lock().unwrap();
        Ok((
            conn.query_row("SELECT count(*) FROM enclosures WHERE state = 'pending'", [], |r| r.get(0))?,
            conn.query_row("SELECT count(*) FROM enclosures WHERE path IS NOT NULL", [], |r| r.get(0))?,
        ))
    }
}


/// An entry plus its enclosures, for the web UI.
#[derive(Debug, serde::Serialize)]
pub struct EntryRow {
    pub guid: String,
    pub feed_id: String,
    pub title: Option<String>,
    pub link: Option<String>,
    pub published: Option<i64>,
    pub description: Option<String>,
    pub read: bool,
    pub flagged: bool,
    pub image: Option<String>,
    pub duration: Option<i64>,
    pub episode: Option<i64>,
    pub season: Option<i64>,
    pub position: i64,
    pub enclosures: Vec<EncRow>,
}

/// The search clause. `?2` is referenced unconditionally -- binding a parameter the
/// statement does not mention is an error, so an empty needle short-circuits instead.
const SEARCH: &str = "(?2 = '' OR lower(coalesce(e.title, '')) LIKE ?2
                      OR lower(coalesce(e.description, '')) LIKE ?2)";

/// Which slice of a feed the UI is asking for.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Filter {
    All,
    Unread,
    Downloaded,
    Flagged,
}

impl Filter {
    pub fn parse(s: &str) -> Self {
        match s {
            "unread" => Self::Unread,
            "downloaded" => Self::Downloaded,
            "flagged" => Self::Flagged,
            _ => Self::All,
        }
    }

    /// The WHERE fragment for this filter. `e` is entries, and the EXISTS subquery is
    /// correlated against it.
    fn sql(self) -> &'static str {
        match self {
            Self::All => "1=1",
            Self::Unread => "e.read = 0",
            Self::Flagged => "e.flagged = 1",
            Self::Downloaded => {
                "EXISTS (SELECT 1 FROM enclosures x
                         WHERE x.feed_id = e.feed_id AND x.guid = e.guid AND x.path IS NOT NULL)"
            }
        }
    }
}

#[derive(Debug, Clone, serde::Serialize)]
pub struct EncRow {
    pub id: i64,
    pub feed_id: String,
    pub guid: String,
    pub url: String,
    pub mime: Option<String>,
    pub length: Option<i64>,
    pub path: Option<String>,
    pub state: String,
    pub last_error: Option<String>,
}

impl Db {
    /// One page of a feed's entries, newest first, each with its enclosures attached.
    /// `search` matches title and description, case-insensitively.
    pub fn entries(
        &self,
        feed_id: &str,
        filter: Filter,
        search: Option<&str>,
        offset: i64,
        limit: i64,
    ) -> Result<Vec<EntryRow>> {
        let conn = self.conn.lock().unwrap();
        let like = search
            .map(|q| format!("%{}%", q.trim().to_lowercase()))
            .unwrap_or_default();
        let sql = format!(
            "SELECT e.guid, e.feed_id, e.title, e.link, e.published, e.description, e.read,
                    e.flagged, e.image, e.duration, e.episode, e.season, e.position
             FROM entries e
             WHERE e.feed_id = ?1 AND {} AND {SEARCH}
             ORDER BY coalesce(e.published, e.first_seen) DESC, e.rowid DESC
             LIMIT ?4 OFFSET ?3",
            filter.sql()
        );
        let mut stmt = conn.prepare(&sql)?;
        let map = |r: &rusqlite::Row| -> rusqlite::Result<EntryRow> {
            Ok(EntryRow {
                guid: r.get(0)?,
                feed_id: r.get(1)?,
                title: r.get(2)?,
                link: r.get(3)?,
                published: r.get(4)?,
                description: r.get(5)?,
                read: r.get::<_, i64>(6)? != 0,
                flagged: r.get::<_, i64>(7)? != 0,
                image: r.get(8)?,
                duration: r.get(9)?,
                episode: r.get(10)?,
                season: r.get(11)?,
                position: r.get(12)?,
                enclosures: vec![],
            })
        };
        let mut rows: Vec<EntryRow> = stmt
            .query_map(rusqlite::params![feed_id, like, offset, limit], map)?
            .collect::<rusqlite::Result<Vec<_>>>()?;

        if rows.is_empty() {
            return Ok(rows);
        }

        // Only the guids on this page, so a feed with thousands of entries stays cheap.
        let placeholders = std::iter::repeat_n("?", rows.len()).collect::<Vec<_>>().join(",");
        let sql = format!(
            "SELECT id, feed_id, guid, url, mime, length, path, state, last_error
             FROM enclosures WHERE feed_id = ? AND guid IN ({placeholders}) ORDER BY id"
        );
        let mut params: Vec<&dyn rusqlite::ToSql> = Vec::with_capacity(rows.len() + 1);
        params.push(&feed_id);
        for row in &rows {
            params.push(&row.guid);
        }
        let mut stmt = conn.prepare(&sql)?;
        let encs = stmt
            .query_map(params.as_slice(), |r| {
                Ok(EncRow {
                    id: r.get(0)?,
                    feed_id: r.get(1)?,
                    guid: r.get(2)?,
                    url: r.get(3)?,
                    mime: r.get(4)?,
                    length: r.get(5)?,
                    path: r.get(6)?,
                    state: r.get(7)?,
                    last_error: r.get(8)?,
                })
            })?
            .collect::<rusqlite::Result<Vec<_>>>()?;

        for enc in encs {
            if let Some(row) = rows.iter_mut().find(|r| r.guid == enc.guid) {
                row.enclosures.push(enc);
            }
        }
        Ok(rows)
    }

    /// How many entries match, so the UI knows whether there is another page.
    pub fn count_entries(&self, feed_id: &str, filter: Filter, search: Option<&str>) -> Result<i64> {
        let conn = self.conn.lock().unwrap();
        let like = search
            .map(|q| format!("%{}%", q.trim().to_lowercase()))
            .unwrap_or_default();
        let sql = format!(
            "SELECT count(*) FROM entries e WHERE e.feed_id = ?1 AND {} AND {SEARCH}",
            filter.sql()
        );
        Ok(conn.query_row(&sql, rusqlite::params![feed_id, like], |r| r.get(0))?)
    }

    /// Where playback got to, so it resumes there next time.
    pub fn set_position(&self, feed_id: &str, guid: &str, secs: i64) -> Result<()> {
        let conn = self.conn.lock().unwrap();
        conn.execute(
            "UPDATE entries SET position = ?3 WHERE feed_id = ?1 AND guid = ?2",
            rusqlite::params![feed_id, guid, secs.max(0)],
        )?;
        Ok(())
    }

    /// Marks every entry in a feed read, for the "mark all read" button.
    /// Marks every entry of the given feeds read. Takes a list because an OPML subscription
    /// holds no entries itself -- marking it read means the feeds inside it.
    pub fn mark_all_read(&self, feed_ids: &[String]) -> Result<usize> {
        let conn = self.conn.lock().unwrap();
        let mut n = 0;
        for id in feed_ids {
            n += conn.execute("UPDATE entries SET read = 1 WHERE feed_id = ?1 AND read = 0", [id])?;
        }
        Ok(n)
    }

    /// The next N enclosures with no file, newest entry first -- what "download latest"
    /// queues up.
    pub fn undownloaded(&self, feed_id: &str, limit: i64) -> Result<Vec<i64>> {
        let conn = self.conn.lock().unwrap();
        let mut stmt = conn.prepare(
            "SELECT x.id FROM enclosures x
             JOIN entries e ON e.feed_id = x.feed_id AND e.guid = x.guid
             WHERE x.feed_id = ?1 AND x.path IS NULL AND x.state != 'reaped'
             ORDER BY coalesce(e.published, e.first_seen) DESC
             LIMIT ?2",
        )?;
        Ok(stmt
            .query_map(rusqlite::params![feed_id, limit], |r| r.get(0))?
            .collect::<rusqlite::Result<Vec<_>>>()?)
    }

    pub fn enclosure(&self, id: i64) -> Result<Option<EncRow>> {
        let conn = self.conn.lock().unwrap();
        Ok(conn
            .query_row(
                "SELECT id, feed_id, guid, url, mime, length, path, state, last_error
                 FROM enclosures WHERE id = ?1",
                [id],
                |r| {
                    Ok(EncRow {
                        id: r.get(0)?,
                        feed_id: r.get(1)?,
                        guid: r.get(2)?,
                        url: r.get(3)?,
                        mime: r.get(4)?,
                        length: r.get(5)?,
                        path: r.get(6)?,
                        state: r.get(7)?,
                        last_error: r.get(8)?,
                    })
                },
            )
            .optional()?)
    }

    /// `read` and `flagged` finally get a writer: retention orders by them.
    pub fn set_entry_flag(&self, feed_id: &str, guid: &str, field: EntryFlag, on: bool) -> Result<()> {
        let conn = self.conn.lock().unwrap();
        let sql = match field {
            EntryFlag::Read => "UPDATE entries SET read = ?3 WHERE feed_id = ?1 AND guid = ?2",
            EntryFlag::Flagged => "UPDATE entries SET flagged = ?3 WHERE feed_id = ?1 AND guid = ?2",
        };
        conn.execute(sql, rusqlite::params![feed_id, guid, on as i64])?;
        Ok(())
    }

    /// How many files this feed has on disk. Decides whether a feed dropped from an OPML
    /// can be removed or must be kept.
    pub fn downloaded_count(&self, feed_id: &str) -> Result<i64> {
        let conn = self.conn.lock().unwrap();
        Ok(conn.query_row(
            "SELECT count(*) FROM enclosures WHERE feed_id = ?1 AND path IS NOT NULL",
            [feed_id],
            |r| r.get(0),
        )?)
    }

    /// Names a feed without touching its conditional-GET validators. An OPML subscription
    /// takes its name from the document's own <head><title>.
    pub fn set_title(&self, feed_id: &str, title: &str) -> Result<()> {
        let conn = self.conn.lock().unwrap();
        conn.execute(
            "UPDATE feeds SET title = ?2 WHERE id = ?1 AND coalesce(title, '') != ?2",
            rusqlite::params![feed_id, title],
        )?;
        Ok(())
    }

    /// Records a feed that came from an OPML. Its settings are the parent's; only what
    /// identifies it is stored.
    pub fn upsert_managed(
        &self,
        id: &str,
        url: &str,
        title: &str,
        group_id: &str,
    ) -> Result<()> {
        let conn = self.conn.lock().unwrap();
        conn.execute(
            "INSERT INTO feeds (id, url, title, group_id, managed, orphaned)
             VALUES (?1, ?2, ?3, ?4, 1, 0)
             ON CONFLICT(id) DO UPDATE SET
                url = excluded.url,
                title = coalesce(feeds.title, excluded.title),
                group_id = excluded.group_id,
                managed = 1,
                orphaned = 0",
            rusqlite::params![id, url, title, group_id],
        )?;
        Ok(())
    }

    /// Every feed derived from an OPML, whichever group.
    pub fn managed_feeds(&self) -> Result<Vec<Managed>> {
        let conn = self.conn.lock().unwrap();
        let mut stmt = conn.prepare(
            "SELECT id, url, title, group_id, orphaned FROM feeds
             WHERE managed = 1 AND group_id IS NOT NULL ORDER BY coalesce(title, id)",
        )?;
        Ok(stmt
            .query_map([], |r| {
                Ok(Managed {
                    id: r.get(0)?,
                    url: r.get(1)?,
                    title: r.get(2)?,
                    group_id: r.get(3)?,
                    orphaned: r.get::<_, i64>(4)? != 0,
                })
            })?
            .collect::<rusqlite::Result<Vec<_>>>()?)
    }

    /// Forgets a derived feed entirely. Only for one with nothing downloaded.
    pub fn drop_managed(&self, id: &str) -> Result<()> {
        let conn = self.conn.lock().unwrap();
        conn.execute("DELETE FROM feeds WHERE id = ?1 AND managed = 1", [id])?;
        conn.execute("DELETE FROM entries WHERE feed_id = ?1", [id])?;
        conn.execute("DELETE FROM enclosures WHERE feed_id = ?1 AND path IS NULL", [id])?;
        Ok(())
    }

    /// Stops treating a feed as derived, because it now has its own config entry.
    pub fn unmanage(&self, id: &str) -> Result<()> {
        let conn = self.conn.lock().unwrap();
        conn.execute("UPDATE feeds SET managed = 0 WHERE id = ?1", [id])?;
        Ok(())
    }

    pub fn set_orphaned(&self, feed_id: &str, on: bool) -> Result<()> {
        let conn = self.conn.lock().unwrap();
        conn.execute(
            "INSERT INTO feeds (id, url, orphaned) VALUES (?1, '', ?2)
             ON CONFLICT(id) DO UPDATE SET orphaned = excluded.orphaned",
            rusqlite::params![feed_id, on as i64],
        )?;
        Ok(())
    }

    /// Nothing can be in flight the moment the daemon starts, so any row still marked
    /// `downloading` is a leftover from a restart or a crash. Left alone it would sit
    /// there forever: the pending queue skips it and nothing else ever revisits it.
    pub fn requeue_interrupted(&self) -> Result<usize> {
        let conn = self.conn.lock().unwrap();
        Ok(conn.execute(
            "UPDATE enclosures SET state = 'pending' WHERE state = 'downloading' AND path IS NULL",
            [],
        )?)
    }

    /// Puts an enclosure back in the queue so the next scan picks it up. This is how a
    /// `skipped` verdict (from a filter that has since been changed) gets revisited.
    pub fn requeue(&self, id: i64) -> Result<()> {
        let conn = self.conn.lock().unwrap();
        conn.execute(
            "UPDATE enclosures SET state = 'pending', last_error = NULL
             WHERE id = ?1 AND path IS NULL",
            [id],
        )?;
        Ok(())
    }
}

#[derive(Debug, Clone, Copy)]
pub enum EntryFlag {
    Read,
    Flagged,
}

/// Unix seconds. Everything time-shaped in the DB is stored this way.
pub fn now() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs() as i64)
        .unwrap_or(0)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn schema_is_idempotent_and_summary_handles_unknown_feeds() {
        let db = Db::memory().unwrap();
        // Re-running the schema must not fail: open() does this on every start.
        db.conn.lock().unwrap().execute_batch(SCHEMA).unwrap();

        let sum = db.feed_summary("never-seen").unwrap();
        assert_eq!(sum.last_checked, None);
        assert_eq!(sum.entries, 0);
        assert_eq!(sum.downloaded, 0);
    }

    #[test]
    fn every_filter_works_with_and_without_a_search_term() {
        // Regression: the search clause used to be omitted when no term was given, while
        // ?2 was still bound -- rusqlite rejects a parameter the statement never mentions,
        // so plain filtering failed with "Wrong number of parameters passed to query".
        let db = Db::memory().unwrap();
        db.exec_for_test(
            "INSERT INTO entries (feed_id, guid, title, description, first_seen, read, flagged) VALUES
                ('f','a','Alpha dive','notes one',100,0,0),
                ('f','b','Beta',      'notes two',200,1,0),
                ('f','c','Gamma dive','notes three',300,1,1);
             INSERT INTO enclosures (id, feed_id, guid, url, path, state) VALUES
                (1,'f','b','u1','/tmp/b','done');",
        )
        .unwrap();

        for f in [Filter::All, Filter::Unread, Filter::Downloaded, Filter::Flagged] {
            // Both paths must run without erroring, and agree with each other.
            let rows = db.entries("f", f, None, 0, 50).unwrap();
            let n = db.count_entries("f", f, None).unwrap();
            assert_eq!(rows.len() as i64, n, "{f:?} count disagrees with the page");

            let rows = db.entries("f", f, Some("dive"), 0, 50).unwrap();
            let n = db.count_entries("f", f, Some("dive")).unwrap();
            assert_eq!(rows.len() as i64, n, "{f:?} with search disagrees");
        }

        assert_eq!(db.count_entries("f", Filter::All, None).unwrap(), 3);
        assert_eq!(db.count_entries("f", Filter::Unread, None).unwrap(), 1);
        assert_eq!(db.count_entries("f", Filter::Downloaded, None).unwrap(), 1);
        assert_eq!(db.count_entries("f", Filter::Flagged, None).unwrap(), 1);
        assert_eq!(db.count_entries("f", Filter::All, Some("dive")).unwrap(), 2);
        assert_eq!(db.count_entries("f", Filter::All, Some("NOTES two")).unwrap(), 1,
            "search is case-insensitive and covers the description");
    }

    #[test]
    fn pending_takes_the_latest_episodes_first() {
        // A cap of 3 must mean the three newest, not the three recorded first.
        let db = Db::memory().unwrap();
        db.exec_for_test(
            "INSERT INTO entries (feed_id, guid, published, first_seen) VALUES
                ('f','old',100,100), ('f','mid',200,200), ('f','new',300,300);
             INSERT INTO enclosures (id, feed_id, guid, url, state) VALUES
                (1,'f','old','u-old','pending'),
                (2,'f','mid','u-mid','pending'),
                (3,'f','new','u-new','pending');",
        )
        .unwrap();
        let got: Vec<String> = db.pending("f", 2).unwrap().into_iter().map(|p| p.url).collect();
        assert_eq!(got, vec!["u-new", "u-mid"], "newest first, oldest left for later");
    }

    #[test]
    fn a_restart_requeues_interrupted_downloads() {
        let db = Db::memory().unwrap();
        db.exec_for_test(
            "INSERT INTO enclosures (id, feed_id, guid, url, state, path) VALUES
                (1,'f','a','u1','downloading',NULL),
                (2,'f','b','u2','pending',NULL),
                (3,'f','c','u3','downloading','/tmp/already-here'),
                (4,'f','d','u4','done','/tmp/x');",
        )
        .unwrap();
        assert_eq!(db.requeue_interrupted().unwrap(), 1, "only the in-flight, fileless one");
        let conn = db.conn.lock().unwrap();
        let state = |id: i64| -> String {
            conn.query_row("SELECT state FROM enclosures WHERE id = ?1", [id], |r| r.get(0)).unwrap()
        };
        assert_eq!(state(1), "pending");
        assert_eq!(state(3), "downloading", "it has a file; leave it alone");
        assert_eq!(state(4), "done");
    }

    #[test]
    fn enclosure_url_is_the_dedupe_key() {
        let db = Db::memory().unwrap();
        let conn = db.conn.lock().unwrap();
        let insert = "INSERT INTO enclosures (feed_id, guid, url, state) VALUES ('f', 'g', 'http://x/a.mp3', 'pending')";
        conn.execute(insert, []).unwrap();
        assert!(conn.execute(insert, []).is_err(), "duplicate url must be rejected");
    }
}