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 <noreply@anthropic.com>

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<i64>,
 }
 
+/// A subscription's keywords, stored as a JSON array.
+fn keywords(json: Option<String>) -> Option<Vec<String>> {
+    json.and_then(|j| serde_json::from_str(&j).ok())
+}
+
+impl From<subscriptions::Model> for Sub {
+    fn from(s: subscriptions::Model) -> Self {
+        Sub {
+            feed_id: s.feed_id,
+            keywords: keywords(s.keywords),
+            auto_download: s.auto_download,
+            allow_explicit: s.allow_explicit,
+            max_new_per_check: s.max_new_per_check,
+        }
+    }
+}
+
 impl From<users::Model> for User {
     fn from(u: users::Model) -> Self {
         User {
@@ -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<sea_orm::Value>) -> Statement {
+        Statement::from_sql_and_values(self.orm.get_database_backend(), sql, values)
+    }
+
+    /// Runs a statement; how many rows it changed.
+    async fn exec(&self, sql: &str, values: Vec<sea_orm::Value>) -> Result<u64> {
+        Ok(self.orm.execute_raw(self.stmt(sql, values)).await?.rows_affected())
+    }
+
+    async fn rows(&self, sql: &str, values: Vec<sea_orm::Value>) -> Result<Vec<sea_orm::QueryResult>> {
+        Ok(self.orm.query_all_raw(self.stmt(sql, values)).await?)
+    }
+
     /// The first admin starts subscribed to the whole catalogue: whoever wrote config.toml meant
     /// to read those feeds, and without this a fresh install signs in to an empty sidebar. Runs
     /// only while nobody subscribes to anything, so an unsubscribe is never undone.
-    pub fn adopt_catalogue(&self, user_id: i64, catalogue: &[String]) -> Result<usize> {
-        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<usize> {
+        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<Option<Sub>> {
-        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<String>>(0)?
-                    .and_then(|j| serde_json::from_str(&j).ok()),
-                auto_download: r.get::<_, Option<i64>>(1)?.map(|v| v != 0),
-                allow_explicit: r.get::<_, Option<i64>>(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<Option<Sub>> {
+        Ok(subscriptions::Entity::find_by_id((user_id, feed_id.to_owned()))
+            .one(&self.orm)
+            .await?
+            .map(Sub::from))
     }
 
     /// Every feed this person subscribes to, with their settings.
-    pub fn subscriptions_for(&self, user_id: i64) -> Result<Vec<Sub>> {
-        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<String>>(1)?
-                        .and_then(|j| serde_json::from_str(&j).ok()),
-                    auto_download: r.get::<_, Option<i64>>(2)?.map(|v| v != 0),
-                    allow_explicit: r.get::<_, Option<i64>>(3)?.map(|v| v != 0),
-                    max_new_per_check: r.get(4)?,
-                })
-            })?
-            .collect::<rusqlite::Result<Vec<_>>>()?;
-        Ok(out)
+    pub async fn subscriptions_for(&self, user_id: i64) -> Result<Vec<Sub>> {
+        Ok(subscriptions::Entity::find()
+            .filter(subscriptions::Column::UserId.eq(user_id))
+            .all(&self.orm)
+            .await?
+            .into_iter()
+            .map(Sub::from)
+            .collect())
     }
 
     /// Everyone's settings for one feed. The scanner merges these into what it fetches
     /// and downloads, since one file serves the lot.
     /// In a group, whatever someone has not set on the feed itself comes from their
     /// subscription to the group, as the group's settings dialog has always said it does.
-    pub fn subscribers(&self, feed_id: &str, group: Option<&str>) -> Result<Vec<Sub>> {
-        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)
+    pub async fn subscribers(&self, feed_id: &str, group: Option<&str>) -> Result<Vec<Sub>> {
+        let rows = self
+            .rows(
+                "SELECT coalesce(c.keywords, p.keywords) AS keywords,
+                        coalesce(c.auto_download, p.auto_download) AS auto_download,
+                        coalesce(c.allow_explicit, p.allow_explicit) AS allow_explicit,
+                        coalesce(c.max_new_per_check, p.max_new_per_check) AS max_new_per_check
                    FROM subscriptions c
-               LEFT JOIN subscriptions p ON p.user_id = c.user_id AND p.feed_id = ?2
-              WHERE c.feed_id = ?1",
-        )?;
-        let out = stmt
-            .query_map(params![feed_id, group], |r| {
+                   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<String>>(0)?
-                        .and_then(|j| serde_json::from_str(&j).ok()),
-                    auto_download: r.get::<_, Option<i64>>(1)?.map(|v| v != 0),
-                    allow_explicit: r.get::<_, Option<i64>>(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::<rusqlite::Result<Vec<_>>>()?;
-        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<std::collections::HashMap<String, i64>> {
-        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::<rusqlite::Result<std::collections::HashMap<_, _>>>()?;
-        Ok(out)
+    pub async fn subscriber_counts(&self) -> Result<std::collections::HashMap<String, i64>> {
+        self.rows("SELECT feed_id, count(*) AS n FROM subscriptions GROUP BY feed_id", vec![])
+            .await?
+            .iter()
+            .map(|r| Ok((r.try_get("", "feed_id")?, r.try_get("", "n")?)))
+            .collect()
     }
 
     /// Feeds with any audio or video enclosure: the Directory's Podcasts, with the rest Blogs.
     /// Reaped files keep their rows, so a show whose files have all been purged still counts.
-    pub fn media_feeds(&self) -> Result<std::collections::HashSet<String>> {
-        let conn = self.conn.lock().unwrap();
-        let mut stmt = conn.prepare(
+    pub async fn media_feeds(&self) -> Result<std::collections::HashSet<String>> {
+        self.rows(
             "SELECT DISTINCT feed_id FROM enclosures WHERE mime LIKE 'audio/%' OR mime LIKE 'video/%'",
-        )?;
-        let out = stmt.query_map([], |r| r.get(0))?.collect::<rusqlite::Result<_>>()?;
-        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)
+    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
+                   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<i64>>(0)?.unwrap_or(0), r.get::<_, Option<i64>>(1)?.unwrap_or(0)))
-        })?;
-        Ok((starred, unread))
+                  WHERE e.id = $1",
+                vec![enclosure_id.into(), user_id.into()],
+            )
+            .await?;
+        let r = r.first().context("an aggregate always returns a row")?;
+        Ok((
+            r.try_get::<Option<i64>>("", "starred")?.unwrap_or(0),
+            r.try_get::<Option<i64>>("", "unread")?.unwrap_or(0),
+        ))
     }
 
-    pub 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<std::collections::HashSet<String>> {
-        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::<rusqlite::Result<_>>()?;
-        Ok(ids)
+    pub async fn pinned_feeds(&self, user_id: i64) -> Result<std::collections::HashSet<String>> {
+        Ok(subscriptions::Entity::find()
+            .filter(subscriptions::Column::UserId.eq(user_id))
+            .filter(subscriptions::Column::Pinned.eq(true))
+            .all(&self.orm)
+            .await?
+            .into_iter()
+            .map(|s| s.feed_id)
+            .collect())
     }
 
     /// False when they do not subscribe to it, since there is then no row in their list to pin.
-    pub fn set_pinned(&self, user_id: i64, feed_id: &str, on: bool) -> Result<bool> {
-        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<bool> {
+        let r = subscriptions::Entity::update_many()
+            .col_expr(subscriptions::Column::Pinned, Expr::val(on).into())
+            .filter(subscriptions::Column::UserId.eq(user_id))
+            .filter(subscriptions::Column::FeedId.eq(feed_id))
+            .exec(&self.orm)
+            .await?;
+        Ok(r.rows_affected > 0)
     }
 
-    pub 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<Option<bool>> {
+        let explicit = async |group| -> Vec<Option<bool>> {
             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]

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<String> = 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 <name> --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<Policy> {
+async fn policy_for(ctx: &Ctx, id: &str, feed_cfg: &config::Feed) -> Result<Policy> {
     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 {

src/web.rs

@@ -676,12 +676,12 @@ async fn feeds(
     let mine: std::collections::HashMap<String, crate::db::Sub> = 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<Vec<PopularRow>> {
+async fn popular(state: &WebState, user_id: i64) -> Result<Vec<PopularRow>> {
     let db = &state.ctx.db;
     let mine: std::collections::HashSet<String> =
-        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<WebState>,
     user: crate::db::User,
 ) -> Result<Json<Vec<PopularRow>>, 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<WebState>,
     user: crate::db::User,
 ) -> Result<Json<Vec<PopularRow>>, 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<String>,
 ) -> Result<Json<serde_json::Value>, 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<StatusCode, ApiError> {
     // 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<StatusCode, ApiError> {
     // 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<Json<serde_json::Value>, ApiError> {
     let ids: Vec<String> =
-        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<String> =
-        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;
     }