From d7f8f2df1d4a8d42704e33bbaf867098a9654a17 Mon Sep 17 00:00:00 2001 From: rays Date: Fri, 18 Sep 2026 18:23:37 +0000 Subject: [PATCH] SeaORM: subscriptions and pins Twelve subscription functions move to SeaORM. Lookups use the entity API; the joins, counts and upserts are SQL written to run on both databases: $n parameters, ON CONFLICT DO NOTHING in place of INSERT OR IGNORE, and CASE WHEN on the yes/no column itself rather than comparing it to 1, which Postgres would refuse for a boolean. INSERT ... SELECT ... ON CONFLICT gets a WHERE true, which SQLite needs to tell the two apart. Checked with a daemon on a copy of production: the feed list, read through the new code, comes back with every feed and its settings. Co-Authored-By: Claude Opus 5 --- src/db.rs | 347 +++++++++++++++++++++++++++------------------------- src/main.rs | 20 +-- src/web.rs | 58 ++++----- 3 files changed, 219 insertions(+), 206 deletions(-) diff --git a/src/db.rs b/src/db.rs index 9fbb9bc..a4034ee 100644 --- a/src/db.rs +++ b/src/db.rs @@ -1,10 +1,13 @@ //! SQLite state. Replaces the per-feed .ipxd plists, history.dat and qmcache.dat. use anyhow::{Context, Result}; -use crate::entity::{sessions, users}; +use crate::entity::{sessions, subscriptions, users}; use rusqlite::{Connection, OptionalExtension, params}; use sea_orm::sea_query::{Expr, Func}; -use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QueryOrder, Set}; +use sea_orm::{ + ActiveModelTrait, ColumnTrait, ConnectionTrait, EntityTrait, PaginatorTrait, QueryFilter, QueryOrder, Set, + Statement, +}; use std::path::Path; use std::sync::Mutex; @@ -203,6 +206,23 @@ pub struct User { pub last_login: Option, } +/// A subscription's keywords, stored as a JSON array. +fn keywords(json: Option) -> Option> { + json.and_then(|j| serde_json::from_str(&j).ok()) +} + +impl From 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 for User { fn from(u: users::Model) -> Self { User { @@ -986,212 +1006,205 @@ impl Db { Ok(()) } + // ---- hand-written SQL ---- + // + // For what reads better as SQL than as a query builder: joins, sums, upserts. Written to + // run on SQLite and Postgres alike -- $1-style parameters, ON CONFLICT, CASE WHEN on a + // yes/no column rather than comparing it to 1 -- since both run it (issue #18). + + fn stmt(&self, sql: &str, values: Vec) -> 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) -> Result { + Ok(self.orm.execute_raw(self.stmt(sql, values)).await?.rows_affected()) + } + + async fn rows(&self, sql: &str, values: Vec) -> Result> { + 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 fn adopt_catalogue(&self, user_id: i64, catalogue: &[String]) -> Result { - let conn = self.conn.lock().unwrap(); - let already: i64 = - conn.query_row("SELECT count(*) FROM subscriptions", [], |r| r.get(0))?; - if already > 0 { + pub async fn adopt_catalogue(&self, user_id: i64, catalogue: &[String]) -> Result { + if subscriptions::Entity::find().count(&self.orm).await? > 0 { return Ok(0); } for id in catalogue { - conn.execute( - "INSERT OR IGNORE INTO subscriptions (user_id, feed_id) VALUES (?1, ?2)", - params![user_id, id], - )?; + self.subscribe(user_id, id).await?; } - // Feeds that exist only in the database (OPML children) count too. - conn.execute( - "INSERT OR IGNORE INTO subscriptions (user_id, feed_id) SELECT ?1, id FROM feeds", - params![user_id], - )?; + // 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 fn subscription(&self, user_id: i64, feed_id: &str) -> Result> { - let conn = self.conn.lock().unwrap(); - let mut stmt = conn.prepare( - "SELECT keywords, auto_download, allow_explicit, max_new_per_check - FROM subscriptions WHERE user_id = ?1 AND feed_id = ?2", - )?; - let mut rows = stmt.query(params![user_id, feed_id])?; - Ok(match rows.next()? { - Some(r) => Some(Sub { - feed_id: feed_id.to_string(), - keywords: r - .get::<_, Option>(0)? - .and_then(|j| serde_json::from_str(&j).ok()), - auto_download: r.get::<_, Option>(1)?.map(|v| v != 0), - allow_explicit: r.get::<_, Option>(2)?.map(|v| v != 0), - max_new_per_check: r.get(3)?, - }), - None => None, - }) + pub async fn subscription(&self, user_id: i64, feed_id: &str) -> Result> { + 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 fn subscriptions_for(&self, user_id: i64) -> Result> { - let conn = self.conn.lock().unwrap(); - let mut stmt = conn.prepare( - "SELECT feed_id, keywords, auto_download, allow_explicit, max_new_per_check - FROM subscriptions WHERE user_id = ?1", - )?; - let out = stmt - .query_map([user_id], |r| { - Ok(Sub { - feed_id: r.get(0)?, - keywords: r - .get::<_, Option>(1)? - .and_then(|j| serde_json::from_str(&j).ok()), - auto_download: r.get::<_, Option>(2)?.map(|v| v != 0), - allow_explicit: r.get::<_, Option>(3)?.map(|v| v != 0), - max_new_per_check: r.get(4)?, - }) - })? - .collect::>>()?; - Ok(out) + pub async fn subscriptions_for(&self, user_id: i64) -> Result> { + 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 fn subscribers(&self, feed_id: &str, group: Option<&str>) -> Result> { - let conn = self.conn.lock().unwrap(); - let mut stmt = conn.prepare( - "SELECT coalesce(c.keywords, p.keywords), coalesce(c.auto_download, p.auto_download), - coalesce(c.allow_explicit, p.allow_explicit), - coalesce(c.max_new_per_check, p.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", - )?; - let out = stmt - .query_map(params![feed_id, group], |r| { + pub async fn subscribers(&self, feed_id: &str, group: Option<&str>) -> Result> { + 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_string(), - keywords: r - .get::<_, Option>(0)? - .and_then(|j| serde_json::from_str(&j).ok()), - auto_download: r.get::<_, Option>(1)?.map(|v| v != 0), - allow_explicit: r.get::<_, Option>(2)?.map(|v| v != 0), - max_new_per_check: r.get(3)?, + 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::>>()?; - Ok(out) + }) + .collect() } /// Subscribers per feed, for the whole catalogue in one query -- the feed list would /// otherwise ask once per feed. - pub fn subscriber_counts(&self) -> Result> { - let conn = self.conn.lock().unwrap(); - let mut stmt = - conn.prepare("SELECT feed_id, count(*) FROM subscriptions GROUP BY feed_id")?; - let out = stmt - .query_map([], |r| Ok((r.get::<_, String>(0)?, r.get::<_, i64>(1)?)))? - .collect::>>()?; - Ok(out) + pub async fn subscriber_counts(&self) -> Result> { + 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 fn media_feeds(&self) -> Result> { - let conn = self.conn.lock().unwrap(); - let mut stmt = conn.prepare( + pub async fn media_feeds(&self) -> Result> { + self.rows( "SELECT DISTINCT feed_id FROM enclosures WHERE mime LIKE 'audio/%' OR mime LIKE 'video/%'", - )?; - let out = stmt.query_map([], |r| r.get(0))?.collect::>()?; - Ok(out) + 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 fn others_wanting(&self, enclosure_id: i64, user_id: i64) -> Result<(i64, i64)> { - let conn = self.conn.lock().unwrap(); - let mut stmt = conn.prepare( - "SELECT - sum(CASE WHEN coalesce(st.flagged, 0) = 1 THEN 1 ELSE 0 END), - sum(CASE WHEN coalesce(st.read, 0) = 0 THEN 1 ELSE 0 END) - 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", - )?; - let (starred, unread) = stmt.query_row(params![enclosure_id, user_id], |r| { - Ok((r.get::<_, Option>(0)?.unwrap_or(0), r.get::<_, Option>(1)?.unwrap_or(0))) - })?; - Ok((starred, unread)) + 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::>("", "starred")?.unwrap_or(0), + r.try_get::>("", "unread")?.unwrap_or(0), + )) } - pub fn subscribe(&self, user_id: i64, feed_id: &str) -> Result<()> { - let conn = self.conn.lock().unwrap(); - conn.execute( - "INSERT OR IGNORE INTO subscriptions (user_id, feed_id) VALUES (?1, ?2)", - params![user_id, feed_id], - )?; + 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 fn pinned_feeds(&self, user_id: i64) -> Result> { - let conn = self.conn.lock().unwrap(); - let mut stmt = conn.prepare("SELECT feed_id FROM subscriptions WHERE user_id = ?1 AND pinned")?; - let ids = stmt.query_map([user_id], |r| r.get(0))?.collect::>()?; - Ok(ids) + pub async fn pinned_feeds(&self, user_id: i64) -> Result> { + 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 fn set_pinned(&self, user_id: i64, feed_id: &str, on: bool) -> Result { - let conn = self.conn.lock().unwrap(); - Ok(conn.execute( - "UPDATE subscriptions SET pinned = ?3 WHERE user_id = ?1 AND feed_id = ?2", - params![user_id, feed_id, on as i64], - )? > 0) + pub async fn set_pinned(&self, user_id: i64, feed_id: &str, on: bool) -> Result { + 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 fn unsubscribe(&self, user_id: i64, feed_id: &str) -> Result<()> { - let conn = self.conn.lock().unwrap(); - conn.execute( - "DELETE FROM subscriptions WHERE user_id = ?1 AND feed_id = ?2", - params![user_id, feed_id], - )?; + 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. - pub fn set_subscription(&self, user_id: i64, sub: &Sub) -> Result<()> { - let conn = self.conn.lock().unwrap(); - let kw = sub - .keywords - .as_ref() - .map(|k| serde_json::to_string(k)) - .transpose()?; - conn.execute( + /// 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 + 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", - params![ - user_id, - sub.feed_id, - kw, - sub.auto_download.map(|v| v as i64), - sub.allow_explicit.map(|v| v as i64), - sub.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(()) } @@ -1796,17 +1809,17 @@ mod tests { .unwrap(); // Nobody has touched it: both others still have it unplayed. - assert_eq!(db.others_wanting(1, 1).unwrap(), (0, 2)); + 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(); - assert_eq!(db.others_wanting(1, 1).unwrap(), (1, 1), "one starred it, one has not played it"); + 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(); - assert_eq!(db.others_wanting(1, 3).unwrap(), (0, 0)); + assert_eq!(db.others_wanting(1, 3).await.unwrap(), (0, 0)); } #[tokio::test] @@ -1917,12 +1930,12 @@ mod tests { 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").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).unwrap().len(), 1, "and still subscribed"); + 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"); } @@ -1955,11 +1968,11 @@ mod tests { db.conn.lock().unwrap().query_row("SELECT count(*) FROM subscriptions", [], |r| r.get(0)).unwrap() }; let catalogue = ["a".to_string(), "b".to_string()]; - assert_eq!(db.adopt_catalogue(1, &catalogue).unwrap(), 2); + assert_eq!(db.adopt_catalogue(1, &catalogue).await.unwrap(), 2); assert_eq!(subs(), 2); // Once anyone subscribes to anything it never runs again, so an unsubscribe sticks. - db.unsubscribe(1, "a").unwrap(); - assert_eq!(db.adopt_catalogue(1, &catalogue).unwrap(), 0); + db.unsubscribe(1, "a").await.unwrap(); + assert_eq!(db.adopt_catalogue(1, &catalogue).await.unwrap(), 0); assert_eq!(subs(), 1); } @@ -2121,14 +2134,14 @@ mod tests { (1,'group',1),(1,'show',NULL),(2,'group',1),(2,'show',0);", ) .unwrap(); - let explicit = |group| -> Vec> { + let explicit = async |group| -> Vec> { let mut v: Vec<_> = - db.subscribers("show", group).unwrap().into_iter().map(|s| s.allow_explicit).collect(); + db.subscribers("show", group).await.unwrap().into_iter().map(|s| s.allow_explicit).collect(); v.sort(); v }; - assert_eq!(explicit(Some("group")), [Some(false), Some(true)], "ray inherits; sam's own choice on the show wins"); - assert_eq!(explicit(None), [None, Some(false)], "outside a group nothing is inherited"); + 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] @@ -2155,15 +2168,15 @@ mod tests { 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).unwrap(), "not subscribed: nothing to pin"); - db.subscribe(me, "f").unwrap(); - assert!(db.set_pinned(me, "f", true).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() }) + db.set_subscription(me, &Sub { feed_id: "f".into(), auto_download: Some(false), ..Default::default() }).await .unwrap(); - assert!(db.pinned_feeds(me).unwrap().contains("f")); - db.set_pinned(me, "f", false).unwrap(); - assert!(db.pinned_feeds(me).unwrap().is_empty()); + assert!(db.pinned_feeds(me).await.unwrap().contains("f")); + db.set_pinned(me, "f", false).await.unwrap(); + assert!(db.pinned_feeds(me).await.unwrap().is_empty()); } #[tokio::test] diff --git a/src/main.rs b/src/main.rs index 764a4a5..a2ccf42 100644 --- a/src/main.rs +++ b/src/main.rs @@ -380,7 +380,7 @@ async fn daemon( if let Some(admin) = ctx.db.users().await?.into_iter().find(|u| u.is_admin) { let catalogue: Vec = ctx.cfg().feeds.keys().cloned().collect(); - match ctx.db.adopt_catalogue(admin.id, &catalogue) { + match ctx.db.adopt_catalogue(admin.id, &catalogue).await { Ok(0) => {} Ok(n) => tracing::info!(user = %admin.name, feeds = n, "subscribed the first admin to the catalogue"), Err(e) => tracing::error!(error = %e, "could not subscribe the first admin to the catalogue"), @@ -676,7 +676,7 @@ async fn import(ctx: &Ctx, config_path: &std::path::Path, file: &std::path::Path .ok_or_else(|| anyhow::anyhow!("no admin account to subscribe: ipx user add --admin"))?; let doc = opml::OPML::from_str(&text) .map_err(|e| anyhow::anyhow!("{} is not OPML: {e}", file.display()))?; - let (added, had) = subscribe_opml(ctx, config_path, &doc, admin.id)?; + let (added, had) = subscribe_opml(ctx, config_path, &doc, admin.id).await?; println!("subscribed {} to {added} feed(s); {had} already there", admin.name); Ok(()) } @@ -691,7 +691,7 @@ async fn import(ctx: &Ctx, config_path: &std::path::Path, file: &std::path::Path /// /// The caller parses the document, so each refuses a file that is not OPML in its own terms, /// before anything is touched: a 400 from the web, a message from the CLI. -pub fn subscribe_opml( +pub async fn subscribe_opml( ctx: &Ctx, config_path: &std::path::Path, doc: &opml::OPML, @@ -746,10 +746,10 @@ pub fn subscribe_opml( let (mut added, mut had) = (0, 0); for id in ids { - if ctx.db.subscription(user_id, &id)?.is_some() { + if ctx.db.subscription(user_id, &id).await?.is_some() { had += 1; } else { - ctx.db.subscribe(user_id, &id)?; + ctx.db.subscribe(user_id, &id).await?; added += 1; } } @@ -1136,7 +1136,7 @@ async fn scan_one( parsed.category.as_deref(), )?; - let policy = policy_for(ctx, id, feed_cfg)?; + let policy = policy_for(ctx, id, feed_cfg).await?; if let Some(parent) = &feed_cfg.group { let listed: Vec<(&str, &str)> = parsed .entries @@ -1318,8 +1318,8 @@ async fn sync_group( .chain(std::iter::once(parent_id.to_string())) { for user in ctx.db.users().await? { - if ctx.db.subscription(user.id, parent_id)?.is_some() { - ctx.db.subscribe(user.id, &id)?; + if ctx.db.subscription(user.id, parent_id).await?.is_some() { + ctx.db.subscribe(user.id, &id).await?; } } } @@ -1404,9 +1404,9 @@ pub struct Policy { pub budget: usize, } -fn policy_for(ctx: &Ctx, id: &str, feed_cfg: &config::Feed) -> Result { +async fn policy_for(ctx: &Ctx, id: &str, feed_cfg: &config::Feed) -> Result { let global = ctx.cfg().general.max_new_per_check; - Ok(merge_policy(&ctx.db.subscribers(id, feed_cfg.group.as_deref())?, feed_cfg, global)) + Ok(merge_policy(&ctx.db.subscribers(id, feed_cfg.group.as_deref()).await?, feed_cfg, global)) } fn merge_policy(subs: &[db::Sub], feed_cfg: &config::Feed, global: usize) -> Policy { diff --git a/src/web.rs b/src/web.rs index 5c4fd72..2f05130 100644 --- a/src/web.rs +++ b/src/web.rs @@ -676,12 +676,12 @@ async fn feeds( let mine: std::collections::HashMap = state .ctx .db - .subscriptions_for(user.id)? + .subscriptions_for(user.id).await? .into_iter() .map(|s| (s.feed_id.clone(), s)) .collect(); - let counts = state.ctx.db.subscriber_counts()?; - let pinned = state.ctx.db.pinned_feeds(user.id)?; + let counts = state.ctx.db.subscriber_counts().await?; + let pinned = state.ctx.db.pinned_feeds(user.id).await?; let mut out = Vec::with_capacity(mine.len()); for sub in &subs { let (id, feed) = (&sub.id, &sub.cfg); @@ -800,12 +800,12 @@ struct PopularRow { /// first. Popular is the top of it, the directory is all of it, and it is all that /// `subscribe_popular` will subscribe you to. An OPML or a Patreon creator is listed as the /// feeds inside it and never itself: both lists are for finding a show. -fn popular(state: &WebState, user_id: i64) -> Result> { +async fn popular(state: &WebState, user_id: i64) -> Result> { let db = &state.ctx.db; let mine: std::collections::HashSet = - db.subscriptions_for(user_id)?.into_iter().map(|s| s.feed_id).collect(); - let counts = db.subscriber_counts()?; - let media = db.media_feeds()?; + db.subscriptions_for(user_id).await?.into_iter().map(|s| s.feed_id).collect(); + let counts = db.subscriber_counts().await?; + let media = db.media_feeds().await?; let catalogue = crate::subscriptions(&state.ctx)?; let by_id: std::collections::HashMap<&str, &crate::config::Feed> = catalogue.iter().map(|s| (s.id.as_str(), &s.cfg)).collect(); @@ -844,7 +844,7 @@ async fn get_popular( State(state): State, user: crate::db::User, ) -> Result>, ApiError> { - let mut rows = popular(&state, user.id)?; + let mut rows = popular(&state, user.id).await?; rows.truncate(10); Ok(Json(rows)) } @@ -854,7 +854,7 @@ async fn get_directory( State(state): State, user: crate::db::User, ) -> Result>, ApiError> { - let mut rows = popular(&state, user.id)?; + let mut rows = popular(&state, user.id).await?; rows.sort_by_key(sort_name); Ok(Json(rows)) } @@ -870,10 +870,10 @@ async fn subscribe_popular( user: crate::db::User, Path(id): Path, ) -> Result, ApiError> { - if !popular(&state, user.id)?.iter().any(|p| p.id == id) { + if !popular(&state, user.id).await?.iter().any(|p| p.id == id) { return Err(ApiError::bad_request(format!("{id:?} is not in the directory"))); } - state.ctx.db.subscribe(user.id, &id)?; + state.ctx.db.subscribe(user.id, &id).await?; Ok(Json(serde_json::json!({ "id": id }))) } @@ -1202,10 +1202,10 @@ struct NewFeed { /// The Add feed dialog's explicit box. Like everything on a feed's own dialog it is yours, so it /// goes on your subscription, and before the first scan, which would otherwise skip every /// explicit item. -fn explicit_on_add(state: &WebState, user_id: i64, feed_id: &str, allow: bool) -> Result<(), ApiError> { +async fn explicit_on_add(state: &WebState, user_id: i64, feed_id: &str, allow: bool) -> Result<(), ApiError> { if allow { let sub = crate::db::Sub { feed_id: feed_id.to_owned(), allow_explicit: Some(true), ..Default::default() }; - state.ctx.db.set_subscription(user_id, &sub)?; + state.ctx.db.set_subscription(user_id, &sub).await?; } Ok(()) } @@ -1223,10 +1223,10 @@ async fn add_feed( .into_iter() .find(|s| crate::feed::same_feed(&s.cfg.url, &url)) { - let already = state.ctx.db.subscription(user.id, &existing.id)?.is_some(); - state.ctx.db.subscribe(user.id, &existing.id)?; + let already = state.ctx.db.subscription(user.id, &existing.id).await?.is_some(); + state.ctx.db.subscribe(user.id, &existing.id).await?; if !already { - explicit_on_add(&state, user.id, &existing.id, body.allow_explicit)?; + explicit_on_add(&state, user.id, &existing.id, body.allow_explicit).await?; } scan_soon(&state, Some(existing.id.clone())).await; return Ok(Json( @@ -1236,8 +1236,8 @@ async fn add_feed( let id = crate::add_one(&state.ctx, &mut cfg, &url, body.folder, body.keywords).await?; cfg.save(&state.config_path)?; state.ctx.reload_cfg(&state.config_path)?; - state.ctx.db.subscribe(user.id, &id)?; - explicit_on_add(&state, user.id, &id, body.allow_explicit)?; + state.ctx.db.subscribe(user.id, &id).await?; + explicit_on_add(&state, user.id, &id, body.allow_explicit).await?; scan_soon(&state, Some(id.clone())).await; Ok(Json(serde_json::json!({ "id": id, "existing": false }))) } @@ -1290,17 +1290,17 @@ async fn patch_feed( ) -> Result { // Pinning is yours alone too, and means nothing for a feed you do not subscribe to. if let Some(on) = body.pinned - && !state.ctx.db.set_pinned(user.id, &id, on)? + && !state.ctx.db.set_pinned(user.id, &id, on).await? { return Err(ApiError::not_found("you do not subscribe to that feed")); } // What one person wants -- which items, whether to fetch them, how many at a time -- // is theirs. It goes on their subscription and nobody else sees the change. - if state.ctx.db.subscription(user.id, &id)?.is_some() { + if state.ctx.db.subscription(user.id, &id).await?.is_some() { let mut mine = state .ctx .db - .subscription(user.id, &id)? + .subscription(user.id, &id).await? .unwrap_or_else(|| crate::db::Sub { feed_id: id.clone(), ..Default::default() }); let mut touched = false; if let Some(v) = body.keywords.clone() { @@ -1320,7 +1320,7 @@ async fn patch_feed( touched = true; } if touched { - state.ctx.db.set_subscription(user.id, &mine)?; + state.ctx.db.set_subscription(user.id, &mine).await?; } } @@ -1400,14 +1400,14 @@ async fn remove_feed( ) -> Result { // Unsubscribing is personal: it takes the feed off your list and leaves everyone // else's alone. - state.ctx.db.unsubscribe(user.id, &id)?; + state.ctx.db.unsubscribe(user.id, &id).await?; for child in crate::subscriptions(&state.ctx)? .iter() .filter(|s| s.cfg.group.as_deref() == Some(id.as_str())) { - state.ctx.db.unsubscribe(user.id, &child.id)?; + state.ctx.db.unsubscribe(user.id, &child.id).await?; } - if state.ctx.db.subscriber_counts()?.contains_key(&id) { + if state.ctx.db.subscriber_counts().await?.contains_key(&id) { return Ok(StatusCode::NO_CONTENT); } @@ -1493,7 +1493,7 @@ async fn delete_file( // There is one copy of the file: deleting it deletes everyone's. Say so before doing // it, once, and let them decide. if !q.force { - let (starred, unread) = state.ctx.db.others_wanting(id, user.id)?; + let (starred, unread) = state.ctx.db.others_wanting(id, user.id).await?; let people = |n: i64| if n == 1 { "person".to_string() } else { format!("{n} people") }; let complaint = match (starred, unread) { (0, 0) => None, @@ -1626,7 +1626,7 @@ async fn read_all_mine( user: crate::db::User, ) -> Result, ApiError> { let ids: Vec = - state.ctx.db.subscriptions_for(user.id)?.into_iter().map(|s| s.feed_id).collect(); + 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)?; Ok(Json(serde_json::json!({ "marked": n }))) } @@ -1668,7 +1668,7 @@ async fn export_opml( // Yours, not the whole catalogue: other people's feeds, and any private URLs in them, are // not yours to download. This used to export config.toml to whoever asked. let mine: std::collections::HashSet = - state.ctx.db.subscriptions_for(user.id)?.into_iter().map(|s| s.feed_id).collect(); + state.ctx.db.subscriptions_for(user.id).await?.into_iter().map(|s| s.feed_id).collect(); let mut doc = opml::OPML { head: Some(opml::Head { title: Some("ipx subscriptions".into()), @@ -1718,7 +1718,7 @@ async fn import_opml( // file arrives as text, is read here, and is gone when the request ends. let doc = opml::OPML::from_str(&body.xml) .map_err(|e| ApiError::bad_request(format!("that is not an OPML file: {e}")))?; - let (added, already) = crate::subscribe_opml(&state.ctx, &state.config_path, &doc, user.id)?; + let (added, already) = crate::subscribe_opml(&state.ctx, &state.config_path, &doc, user.id).await?; if added > 0 { scan_soon(&state, None).await; }