diff --git a/src/db.rs b/src/db.rs index a4034ee..93b3beb 100644 --- a/src/db.rs +++ b/src/db.rs @@ -1,7 +1,7 @@ //! SQLite state. Replaces the per-feed .ipxd plists, history.dat and qmcache.dat. use anyhow::{Context, Result}; -use crate::entity::{sessions, subscriptions, users}; +use crate::entity::{enclosures, sessions, subscriptions, users}; use rusqlite::{Connection, OptionalExtension, params}; use sea_orm::sea_query::{Expr, Func}; use sea_orm::{ @@ -206,6 +206,22 @@ pub struct User { pub last_login: Option, } +impl From 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) -> Option> { json.and_then(|j| serde_json::from_str(&j).ok()) @@ -722,25 +738,30 @@ impl Db { } impl Db { - pub fn unread_count(&self, user_id: i64, feed_id: &str) -> Result { - let conn = self.conn.lock().unwrap(); - Ok(conn.query_row( - "SELECT count(*) 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 coalesce(s.read, 0) = 0", - rusqlite::params![feed_id, user_id], - |r| r.get(0), - )?) + pub async fn unread_count(&self, user_id: i64, feed_id: &str) -> Result { + 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 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))?, - )) + 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)) } } @@ -764,22 +785,45 @@ pub struct EntryRow { pub enclosures: Vec, } -/// 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)"; +/// 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); -/// Which feeds a query covers: one, or every feed the person subscribes to. Both forms -/// mention `?1`, since binding a parameter the statement does not use is an error. -fn scope_sql(feed_id: Option<&str>, user_param: u8) -> String { - match feed_id { - Some(_) => "e.feed_id = ?1".into(), - None => format!( - "?1 IS NULL AND e.feed_id IN (SELECT feed_id FROM subscriptions WHERE user_id = ?{user_param})" - ), +impl Args { + fn p(&mut self, v: impl Into) -> 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". @@ -791,7 +835,7 @@ fn scope_sql(feed_id: Option<&str>, user_param: u8) -> String { /// where the direction chosen is the point. pub fn order_sql(col: &str, dir: &str, pinned_first: bool) -> String { let expr = match col { - "kept" => "coalesce(s.flagged, 0)", + "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)", @@ -799,8 +843,9 @@ pub fn order_sql(col: &str, dir: &str, pinned_first: bool) -> String { _ => "coalesce(e.published, e.first_seen)", }; let dir = if dir == "asc" { "ASC" } else { "DESC" }; - let pins = if pinned_first && col != "kept" { "coalesce(s.flagged, 0) DESC, " } else { "" }; - format!("{pins}{expr} {dir}, coalesce(e.published, e.first_seen) DESC, e.rowid DESC") + let pins = if pinned_first && col != "kept" { "coalesce(s.flagged, false) DESC, " } else { "" }; + // The guid breaks what ties remain: SQLite's rowid did, and Postgres has none. + format!("{pins}{expr} {dir}, coalesce(e.published, e.first_seen) DESC, e.guid DESC") } /// Which slice of a feed the UI is asking for. @@ -834,9 +879,11 @@ impl Filter { /// correlated against it. fn sql(self) -> &'static str { match self { - Self::All => "1=1", - Self::Unread => "coalesce(s.read, 0) = 0", - Self::Flagged => "coalesce(s.flagged, 0) = 1", + // 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)" @@ -867,7 +914,8 @@ 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. - pub fn entries_in( + #[allow(clippy::too_many_arguments)] + pub async fn entries_in( &self, user_id: i64, feed_id: Option<&str>, @@ -877,108 +925,73 @@ impl Db { limit: i64, order: &str, ) -> Result> { - let conn = self.conn.lock().unwrap(); - let like = search - .map(|q| format!("%{}%", q.trim().to_lowercase())) - .unwrap_or_default(); + 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, 0), coalesce(s.flagged, 0), e.image, - coalesce(s.duration, e.duration), - e.episode, e.season, coalesce(s.position, 0) - FROM entries e - LEFT JOIN entry_state s - ON s.user_id = ?5 AND s.feed_id = e.feed_id AND s.guid = e.guid - WHERE {} AND {} AND {SEARCH} + 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 ?4 OFFSET ?3", - scope_sql(feed_id, 5), - filter.sql() + LIMIT {} OFFSET {}", + a.p(limit), + a.p(offset) ); - let mut stmt = conn.prepare(&sql)?; - let map = |r: &rusqlite::Row| -> rusqlite::Result { - 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 = 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![], + }) }) - }; - let mut rows: Vec = stmt - .query_map(rusqlite::params![feed_id, like, offset, limit, user_id], map)? - .collect::>>()?; - + .collect::>>()?; 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 placeholders = std::iter::repeat_n("?", rows.len()).collect::>().join(","); - let sql = format!( - "SELECT id, feed_id, guid, url, mime, length, path, state, last_error - FROM enclosures WHERE guid IN ({placeholders}) ORDER BY id" - ); - let mut params: Vec<&dyn rusqlite::ToSql> = Vec::with_capacity(rows.len()); - 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::>>()?; - + 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(enc); + 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 fn count_in( + pub async fn count_in( &self, user_id: i64, feed_id: Option<&str>, filter: Filter, search: Option<&str>, ) -> Result { - 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 - LEFT JOIN entry_state s - ON s.user_id = ?3 AND s.feed_id = e.feed_id AND s.guid = e.guid - WHERE {} AND {} AND {SEARCH}", - scope_sql(feed_id, 3), - filter.sql() - ); - Ok(conn.query_row(&sql, rusqlite::params![feed_id, like, user_id], |r| r.get(0))?) + 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. @@ -987,7 +1000,7 @@ impl Db { /// 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 fn set_position( + pub async fn set_position( &self, user_id: i64, feed_id: &str, @@ -995,14 +1008,21 @@ impl Db { secs: i64, duration: Option, ) -> Result<()> { - let conn = self.conn.lock().unwrap(); - conn.execute( + // 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, duration)", - rusqlite::params![user_id, feed_id, guid, secs.max(0), duration.filter(|d| *d > 0)], - )?; + 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(()) } @@ -1361,19 +1381,20 @@ impl Db { /// 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, user_id: i64, feed_ids: &[String]) -> Result { - let conn = self.conn.lock().unwrap(); + pub async fn mark_all_read(&self, user_id: i64, feed_ids: &[String]) -> Result { let mut n = 0; for id in feed_ids { - n += conn.execute( - "INSERT INTO entry_state (user_id, feed_id, guid, read) - SELECT ?1, e.feed_id, e.guid, 1 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 coalesce(s.read, 0) = 0 - ON CONFLICT(user_id, feed_id, guid) DO UPDATE SET read = 1", - rusqlite::params![user_id, id], - )?; + 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) } @@ -1419,7 +1440,7 @@ impl Db { } /// Read and kept, per person. The row is created on first touch. - pub fn set_entry_flag( + pub async fn set_entry_flag( &self, user_id: i64, feed_id: &str, @@ -1427,19 +1448,19 @@ impl Db { field: EntryFlag, on: bool, ) -> Result<()> { - let conn = self.conn.lock().unwrap(); let col = match field { EntryFlag::Read => "read", EntryFlag::Flagged => "flagged", }; - conn.execute( + 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}" + VALUES ($1, $2, $3, $4) + ON CONFLICT (user_id, feed_id, guid) DO UPDATE SET {col} = excluded.{col}" ), - rusqlite::params![user_id, feed_id, guid, on as i64], - )?; + vec![user_id.into(), feed_id.into(), guid.into(), on.into()], + ) + .await?; Ok(()) } @@ -1765,34 +1786,34 @@ mod tests { (3,'f','c','u3','video/mp4',2000,'pending');", ) .unwrap(); - let order = |col: &str, dir: &str| -> Vec { - db.entries_in(1, None, Filter::All, None, 0, 50, &order_sql(col, dir, false)) + let order = async |col: &str, dir: &str| -> Vec { + db.entries_in(1, None, Filter::All, None, 0, 50, &order_sql(col, dir, false)).await .unwrap() .into_iter() .map(|e| e.guid) .collect() }; - assert_eq!(order("title", "asc"), ["b", "a", "c"], "Apple, banana, cherry: case folded"); - assert_eq!(order("title", "desc"), ["c", "a", "b"]); - assert_eq!(order("feed", "asc"), ["b", "c", "a"], "Aardvark, then Zebra's newest first"); - assert_eq!(order("type", "asc"), ["a", "b", "c"], "audio, image, video"); - assert_eq!(order("size", "desc"), ["c", "a", "b"]); - assert_eq!(order("published", "desc"), ["c", "b", "a"]); - db.set_entry_flag(1, "f", "a", EntryFlag::Flagged, true).unwrap(); - assert_eq!(order("kept", "desc")[0], "a"); + assert_eq!(order("title", "asc").await, ["b", "a", "c"], "Apple, banana, cherry: case folded"); + assert_eq!(order("title", "desc").await, ["c", "a", "b"]); + assert_eq!(order("feed", "asc").await, ["b", "c", "a"], "Aardvark, then Zebra's newest first"); + assert_eq!(order("type", "asc").await, ["a", "b", "c"], "audio, image, video"); + assert_eq!(order("size", "desc").await, ["c", "a", "b"]); + assert_eq!(order("published", "desc").await, ["c", "b", "a"]); + db.set_entry_flag(1, "f", "a", EntryFlag::Flagged, true).await.unwrap(); + assert_eq!(order("kept", "desc").await[0], "a"); // Pinned first: the pinned banana tops every sort, the rest in the order asked for. - let pinned = |col: &str, dir: &str| -> Vec { - db.entries_in(1, None, Filter::All, None, 0, 50, &order_sql(col, dir, true)) + let pinned = async |col: &str, dir: &str| -> Vec { + db.entries_in(1, None, Filter::All, None, 0, 50, &order_sql(col, dir, true)).await .unwrap() .into_iter() .map(|e| e.guid) .collect() }; - assert_eq!(pinned("published", "desc"), ["a", "c", "b"]); - assert_eq!(pinned("title", "desc"), ["a", "c", "b"]); - assert_eq!(pinned("kept", "asc")[2], "a", "sorting by the pin itself keeps its direction"); + assert_eq!(pinned("published", "desc").await, ["a", "c", "b"]); + assert_eq!(pinned("title", "desc").await, ["a", "c", "b"]); + assert_eq!(pinned("kept", "asc").await[2], "a", "sorting by the pin itself keeps its direction"); // An unknown column or direction is newest first; the name itself never reaches the SQL. - assert_eq!(order("title; DROP TABLE entries", "sideways"), ["c", "b", "a"]); + assert_eq!(order("title; DROP TABLE entries", "sideways").await, ["c", "b", "a"]); assert!(!order_sql("x'; --", "asc", false).contains("x'")); } @@ -1812,13 +1833,13 @@ mod tests { 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).unwrap(); - db.set_entry_flag(3, "f", "a", EntryFlag::Flagged, true).unwrap(); + 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).unwrap(); + db.set_entry_flag(1, "f", "a", EntryFlag::Read, true).await.unwrap(); assert_eq!(db.others_wanting(1, 3).await.unwrap(), (0, 0)); } @@ -1832,19 +1853,19 @@ mod tests { ) .unwrap(); - assert_eq!(db.unread_count(1, "f").unwrap(), 2); - assert_eq!(db.unread_count(2, "f").unwrap(), 2); + 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).unwrap(); - assert_eq!(db.unread_count(1, "f").unwrap(), 1, "ray read one of them"); - assert_eq!(db.unread_count(2, "f").unwrap(), 2, "sam has read nothing"); + 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).unwrap(); - db.set_position(2, "f", "b", 42, Some(600)).unwrap(); + db.set_entry_flag(1, "f", "b", EntryFlag::Flagged, true).await.unwrap(); + db.set_position(2, "f", "b", 42, Some(600)).await.unwrap(); let order = order_sql("published", "desc", false); - let page = |user| db.entries_in(user, Some("f"), Filter::All, None, 0, 50, &order).unwrap(); - let (ray, sam) = (page(1), page(2)); + 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); @@ -1854,9 +1875,9 @@ mod tests { 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()]).unwrap(), 2); - assert_eq!(db.unread_count(2, "f").unwrap(), 0); - assert_eq!(db.unread_count(1, "f").unwrap(), 1); + 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] @@ -2013,42 +2034,42 @@ mod tests { for f in [Filter::All, Filter::Unread, Filter::Downloaded, Filter::Flagged, Filter::InProgress] { // Both paths must run without erroring, and agree with each other. let order = order_sql("published", "desc", false); - let rows = db.entries_in(7, Some("f"), f, None, 0, 50, &order).unwrap(); - let n = db.count_in(7, Some("f"), f, None).unwrap(); + 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).unwrap(); - let n = db.count_in(7, Some("f"), f, Some("dive")).unwrap(); + 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).unwrap(), 7); - assert_eq!(db.count_in(7, Some("f"), Filter::Unread, None).unwrap(), 3); - assert_eq!(db.count_in(7, Some("f"), Filter::Downloaded, None).unwrap(), 1); - assert_eq!(db.count_in(7, Some("f"), Filter::Flagged, None).unwrap(), 1); - assert_eq!(db.count_in(7, Some("f"), Filter::All, Some("dive")).unwrap(), 2); - assert_eq!(db.count_in(7, Some("f"), Filter::All, Some("NOTES two")).unwrap(), 1, + assert_eq!(db.count_in(7, Some("f"), Filter::All, None).await.unwrap(), 7); + assert_eq!(db.count_in(7, Some("f"), Filter::Unread, None).await.unwrap(), 3); + assert_eq!(db.count_in(7, Some("f"), Filter::Downloaded, None).await.unwrap(), 1); + assert_eq!(db.count_in(7, Some("f"), Filter::Flagged, None).await.unwrap(), 1); + assert_eq!(db.count_in(7, Some("f"), Filter::All, Some("dive")).await.unwrap(), 2); + assert_eq!(db.count_in(7, Some("f"), Filter::All, Some("NOTES two")).await.unwrap(), 1, "search is case-insensitive and covers the description"); // Currently Listening: started, not finished, and not just an accidental tap. let order = order_sql("published", "desc", false); - let listening = || { - let rows = db.entries_in(7, Some("f"), Filter::InProgress, None, 0, 50, &order).unwrap(); + 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::>() }; - assert_eq!(listening(), ["h", "d"]); + 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)).unwrap(); - db.set_position(7, "f", "e", 42, Some(9000)).unwrap(); - assert_eq!(listening(), ["h", "e"]); - let rows = db.entries_in(7, Some("f"), Filter::All, None, 0, 50, &order).unwrap(); + 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).unwrap(); - assert_eq!(listening(), ["h", "e"]); + db.set_position(7, "f", "e", 43, None).await.unwrap(); + assert_eq!(listening().await, ["h", "e"]); } #[tokio::test] diff --git a/src/ipc.rs b/src/ipc.rs index 453f404..231cdd1 100644 --- a/src/ipc.rs +++ b/src/ipc.rs @@ -176,7 +176,9 @@ pub async fn daemon_is_live(path: &Path) -> bool { /// healthcheck left waiting behind a scan or a long download timed out and called a busy daemon /// dead. The answer goes to the client that asked and no one else: broadcast, it ended any /// `ipx fetch` that was watching a scan, since `status` is a terminal event. -pub type StatusFn = std::sync::Arc Event + Send + Sync>; +/// A future, since reading the counts is a database query. +pub type StatusFn = + std::sync::Arc std::pin::Pin + Send>> + Send + Sync>; /// Accepts connections, feeding commands to `cmds` and events from `events` back out. pub async fn serve( @@ -249,7 +251,7 @@ async fn handle( // Answered here, not queued behind whatever the worker is on: see StatusFn. Ok(Command::Status) => { tracing::info!(target: "ipx::io", "-> {line}"); - let ev = status(); + let ev = status().await; if let Ok(json) = serde_json::to_string(&ev) { tracing::info!(target: "ipx::io", "<- {json}"); } @@ -366,7 +368,8 @@ mod tests { // Another client, watching a scan: it must not be handed someone else's answer, which // would end its session. let mut watcher = events.subscribe(); - let status: StatusFn = std::sync::Arc::new(|| Event::Status { feeds: 1, pending: 2, downloaded: 3 }); + let status: StatusFn = + std::sync::Arc::new(|| Box::pin(async { Event::Status { feeds: 1, pending: 2, downloaded: 3 } })); let (client, server) = UnixStream::pair().unwrap(); tokio::spawn(handle(server, events.subscribe(), cmds, status)); diff --git a/src/main.rs b/src/main.rs index a2ccf42..56321fb 100644 --- a/src/main.rs +++ b/src/main.rs @@ -340,7 +340,7 @@ async fn run(ctx: &Arc, cmd: Cmd) -> Result<()> { Cmd::Reap { dry_run } => reap(ctx, dry_run, true), Cmd::Download { enclosure } => download_one(ctx, enclosure).await, Cmd::Status => { - ctx.out.emit(status(ctx)); + ctx.out.emit(status(ctx).await); Ok(()) } } @@ -348,8 +348,8 @@ async fn run(ctx: &Arc, cmd: Cmd) -> Result<()> { /// The counts `ipx status` prints. A running daemon's socket answers with this directly rather /// than through the job queue. -fn status(ctx: &Ctx) -> Event { - match ctx.db.counts() { +async fn status(ctx: &Ctx) -> Event { + match ctx.db.counts().await { Ok((pending, downloaded)) => { let feeds = subscriptions(ctx).map(|s| s.len()).unwrap_or(0); Event::Status { feeds, pending, downloaded } @@ -420,7 +420,10 @@ async fn daemon( // status is answered by the socket itself; everything else waits its turn in the queue. let answer: ipc::StatusFn = { let ctx = ctx.clone(); - Arc::new(move || status(&ctx)) + Arc::new(move || { + let ctx = ctx.clone(); + Box::pin(async move { status(&ctx).await }) + }) }; let server = tokio::spawn(ipc::serve(socket.clone(), events.clone(), tx_cmd, answer)); diff --git a/src/web.rs b/src/web.rs index 2f05130..67a36e8 100644 --- a/src/web.rs +++ b/src/web.rs @@ -740,7 +740,7 @@ async fn feeds( last_error: s.last_error, entries: s.entries, downloaded: s.downloaded, - unread: state.ctx.db.unread_count(user.id, id)?, + unread: state.ctx.db.unread_count(user.id, id).await?, subscribers: counts.get(id).copied().unwrap_or(0), pinned: pinned.contains(id), }); @@ -1124,7 +1124,7 @@ async fn entries( user: crate::db::User, Query(page): Query, ) -> Result, ApiError> { - entry_page(&state, user.id, Some(&id), &page) + entry_page(&state, user.id, Some(&id), &page).await } /// Every subscribed feed's items together, newest first: All Subscriptions. @@ -1133,11 +1133,11 @@ async fn all_entries( user: crate::db::User, Query(page): Query, ) -> Result, ApiError> { - entry_page(&state, user.id, None, &page) + entry_page(&state, user.id, None, &page).await } /// One feed's page of items, or every subscribed feed's when `feed` is None. -fn entry_page( +async fn entry_page( state: &WebState, user_id: i64, feed: Option<&str>, @@ -1153,14 +1153,14 @@ fn entry_page( filter != crate::db::Filter::InProgress, ); let mut rows = - db.entries_in(user_id, feed, filter, search, page.offset, page.limit.clamp(1, 200), &order)?; + db.entries_in(user_id, feed, filter, search, page.offset, page.limit.clamp(1, 200), &order).await?; let mut sanitizer = feed_sanitizer(); for row in &mut rows { if let Some(d) = &row.description { row.description = Some(clean_description(&mut sanitizer, d, row.link.as_deref())); } } - let total = db.count_in(user_id, feed, filter, search)?; + let total = db.count_in(user_id, feed, filter, search).await?; Ok(Json(EntryPage { total, entries: rows })) } @@ -1440,10 +1440,10 @@ async fn set_flags( ) -> Result { use crate::db::EntryFlag; if let Some(v) = body.read { - state.ctx.db.set_entry_flag(user.id, &feed_id, &guid, EntryFlag::Read, v)?; + state.ctx.db.set_entry_flag(user.id, &feed_id, &guid, EntryFlag::Read, v).await?; } if let Some(v) = body.flagged { - state.ctx.db.set_entry_flag(user.id, &feed_id, &guid, EntryFlag::Flagged, v)?; + state.ctx.db.set_entry_flag(user.id, &feed_id, &guid, EntryFlag::Flagged, v).await?; } Ok(StatusCode::NO_CONTENT) } @@ -1598,7 +1598,7 @@ async fn set_position( user: crate::db::User, Json(body): Json, ) -> Result { - state.ctx.db.set_position(user.id, &feed_id, &guid, body.secs, body.duration)?; + state.ctx.db.set_position(user.id, &feed_id, &guid, body.secs, body.duration).await?; Ok(StatusCode::NO_CONTENT) } @@ -1615,7 +1615,7 @@ async fn read_all( .filter(|s| s.cfg.group.as_deref() == Some(id.as_str())) .map(|s| s.id), ); - let n = state.ctx.db.mark_all_read(user.id, &ids)?; + let n = state.ctx.db.mark_all_read(user.id, &ids).await?; Ok(Json(serde_json::json!({ "marked": n }))) } @@ -1627,7 +1627,7 @@ async fn read_all_mine( ) -> Result, ApiError> { let ids: Vec = state.ctx.db.subscriptions_for(user.id).await?.into_iter().map(|s| s.feed_id).collect(); - let n = state.ctx.db.mark_all_read(user.id, &ids)?; + let n = state.ctx.db.mark_all_read(user.id, &ids).await?; Ok(Json(serde_json::json!({ "marked": n }))) }