server: fetch search summaries in parallel

server/src/newsreader.rs

@@ -1,6 +1,7 @@
 use std::collections::HashMap;
 
 use cacher::FilesystemCacher;
+use futures::{stream::FuturesUnordered, StreamExt};
 use log::info;
 use maplit::hashmap;
 use scraper::Selector;
@@ -254,23 +255,25 @@ pub async fn set_read_status<'ctx>(
 }
 #[instrument(name = "newsreader::refresh", skip_all)]
 pub async fn refresh<'ctx>(pool: &PgPool, cacher: &FilesystemCacher) -> Result<bool, ServerError> {
-    let body_transformers: Vec<Box<dyn Transformer>> = vec![
-        Box::new(SlurpContents {
-            cacher,
-            inline_css: true,
-            site_selectors: slurp_contents_selectors(),
-        }),
-        Box::new(StripHtml),
-    ];
+    async fn update_search_summary(
+        pool: &PgPool,
+        cacher: &FilesystemCacher,
+        link: Url,
+        body: String,
+        id: i32,
+    ) -> Result<(), ServerError> {
+        let body_transformers: Vec<Box<dyn Transformer>> = vec![
+            Box::new(SlurpContents {
+                cacher,
+                inline_css: true,
+                site_selectors: slurp_contents_selectors(),
+            }),
+            Box::new(StripHtml),
+        ];
 
-    let rows = sqlx::query_file!("sql/need-search-summary.sql",)
-        .fetch_all(pool)
-        .await?;
-    for r in rows {
-        let link = Url::parse(&r.link)?;
         info!("adding {link} to search index");
+        let mut body = body;
         let link = Some(link);
-        let mut body = r.clean_summary.unwrap_or("NO SUMMARY".to_string());
         for t in body_transformers.iter() {
             if t.should_run(&link, &body) {
                 body = t.transform(&link, &body).await?;
@@ -279,10 +282,35 @@ pub async fn refresh<'ctx>(pool: &PgPool, cacher: &FilesystemCacher) -> Result<b
         sqlx::query!(
             "UPDATE post SET search_summary = $1 WHERE id = $2",
             body,
-            r.id
+            id
         )
         .execute(pool)
         .await?;
+        Ok(())
+    }
+
+    let mut unordered: FuturesUnordered<_> = sqlx::query_file!("sql/need-search-summary.sql",)
+        .fetch_all(pool)
+        .await?
+        .into_iter()
+        .map(|r| {
+            let link = Url::parse(&r.link).expect("failed to parse link");
+            let body = r.clean_summary.unwrap_or("NO SUMMARY".to_string());
+            let id = r.id;
+            update_search_summary(pool, cacher, link, body, id)
+        })
+        .collect();
+
+    while let Some(res) = unordered.next().await {
+        //let res = res;
+        match res {
+            Ok(()) => {}
+            Err(err) => {
+                info!("failed refresh {err:?}");
+                // TODO:
+                //fd.error = Some(err);
+            }
+        };
     }
     Ok(true)
 }