Merge news and email search results

server/src/graphql.rs

@@ -1,9 +1,11 @@
 use async_graphql::{
-    connection::Connection, Context, EmptySubscription, Enum, Error, FieldResult, Object, Schema,
+    connection::{self, Connection, Edge, OpaqueCursor},
+    Context, EmptySubscription, Enum, Error, FieldResult, InputObject, Object, Schema,
     SimpleObject, Union,
 };
 use log::info;
 use notmuch::Notmuch;
+use serde::{Deserialize, Serialize};
 use sqlx::postgres::PgPool;
 
 use crate::{newsreader, nm};
@@ -200,6 +202,12 @@ pub struct Tag {
     pub unread: usize,
 }
 
+#[derive(Serialize, Deserialize, Debug, InputObject)]
+struct SearchCursor {
+    newsreader_offset: i32,
+    notmuch_offset: i32,
+}
+
 pub struct QueryRoot;
 #[Object]
 impl QueryRoot {
@@ -207,12 +215,9 @@ impl QueryRoot {
         let nm = ctx.data_unchecked::<Notmuch>();
         let pool = ctx.data_unchecked::<PgPool>();
 
-        // TODO: make this search both copra and merge results
-        if newsreader::is_newsreader_search(&query) {
-            Ok(newsreader::count(pool, &query).await?)
-        } else {
-            Ok(nm::count(nm, &query).await?)
-        }
+        let newsreader_query: newsreader::Query = query.parse()?;
+
+        Ok(newsreader::count(pool, &newsreader_query).await? + nm::count(nm, &query).await?)
     }
 
     async fn search<'ctx>(
@@ -223,17 +228,117 @@ impl QueryRoot {
         first: Option<i32>,
         last: Option<i32>,
         query: String,
-    ) -> Result<Connection<usize, ThreadSummary>, Error> {
-        info!("search({after:?} {before:?} {first:?} {last:?} {query:?})");
+    ) -> Result<Connection<OpaqueCursor<SearchCursor>, ThreadSummary>, Error> {
+        // TODO: add keywords to limit search to one corpus, i.e. is:news or is:mail
+        info!("search({after:?} {before:?} {first:?} {last:?} {query:?})",);
         let nm = ctx.data_unchecked::<Notmuch>();
         let pool = ctx.data_unchecked::<PgPool>();
 
-        // TODO: make this search both copra and merge results
-        if newsreader::is_newsreader_search(&query) {
-            Ok(newsreader::search(pool, after, before, first, last, query).await?)
-        } else {
-            Ok(nm::search(nm, after, before, first, last, query).await?)
+        enum ThreadSummaryCursor {
+            Newsreader(i32, ThreadSummary),
+            Notmuch(i32, ThreadSummary),
         }
+        Ok(connection::query(
+            after,
+            before,
+            first,
+            last,
+            |after: Option<OpaqueCursor<SearchCursor>>,
+             before: Option<OpaqueCursor<SearchCursor>>,
+             first: Option<usize>,
+             last: Option<usize>| async move {
+                info!(
+                    "search({:?} {:?} {first:?} {last:?} {query:?})",
+                    after.as_ref().map(|v| &v.0),
+                    before.as_ref().map(|v| &v.0)
+                );
+                let newsreader_after = after.as_ref().map(|sc| sc.newsreader_offset);
+                let notmuch_after = after.as_ref().map(|sc| sc.newsreader_offset);
+                let newsreader_before = before.as_ref().map(|sc| sc.newsreader_offset);
+                let notmuch_before = before.as_ref().map(|sc| sc.notmuch_offset);
+
+                let newsreader_query: newsreader::Query = query.parse()?;
+                let newsreader_results = newsreader::search(
+                    pool,
+                    newsreader_after,
+                    newsreader_before,
+                    first.map(|v| v as i32),
+                    last.map(|v| v as i32),
+                    &newsreader_query,
+                )
+                .await?
+                .into_iter()
+                .map(|(cur, ts)| ThreadSummaryCursor::Newsreader(cur, ts));
+
+                let notmuch_results = nm::search(
+                    nm,
+                    notmuch_after,
+                    notmuch_before,
+                    first.map(|v| v as i32),
+                    last.map(|v| v as i32),
+                    query,
+                )
+                .await?
+                .into_iter()
+                .map(|(cur, ts)| ThreadSummaryCursor::Notmuch(cur, ts));
+
+                let mut results: Vec<_> = newsreader_results.chain(notmuch_results).collect();
+
+                // The leading '-' is to reverse sort
+                results.sort_by_key(|item| match item {
+                    ThreadSummaryCursor::Newsreader(_, ts) => -ts.timestamp,
+                    ThreadSummaryCursor::Notmuch(_, ts) => -ts.timestamp,
+                });
+
+                let mut has_next_page = before.is_some();
+                if let Some(first) = first {
+                    if results.len() > first {
+                        has_next_page = true;
+                        results.truncate(first);
+                    }
+                }
+
+                let mut has_previous_page = after.is_some();
+                if let Some(last) = last {
+                    if results.len() > last {
+                        has_previous_page = true;
+                        // TODO: find better way to do this.
+                        results.reverse();
+                        results.truncate(last);
+                        results.reverse();
+                    }
+                }
+
+                let mut connection = Connection::new(has_previous_page, has_next_page);
+                let mut newsreader_offset = 0;
+                let mut notmuch_offset = 0;
+
+                connection.edges.extend(results.into_iter().map(|item| {
+                    let thread_summary;
+                    match item {
+                        ThreadSummaryCursor::Newsreader(offset, ts) => {
+                            thread_summary = ts;
+                            newsreader_offset = offset;
+                        }
+                        ThreadSummaryCursor::Notmuch(offset, ts) => {
+                            thread_summary = ts;
+                            notmuch_offset = offset;
+                        }
+                    }
+                    info!(
+                        "item: {} {}",
+                        thread_summary.subject, thread_summary.timestamp
+                    );
+                    let cur = OpaqueCursor(SearchCursor {
+                        newsreader_offset,
+                        notmuch_offset,
+                    });
+                    Edge::new(cur, thread_summary)
+                }));
+                Ok::<_, async_graphql::Error>(connection)
+            },
+        )
+        .await?)
     }
 
     async fn tags<'ctx>(&self, ctx: &Context<'ctx>) -> FieldResult<Vec<Tag>> {