Group search: merge local members on ids, load versions for the page alone

crates/core/src/repo/hosted.rs

@@ -93,8 +93,9 @@ impl HostedRepo {
     /// before paging: the ids that match are listed once, filtered by the
     /// selector, counted and paged, and only the page's versions are loaded.
     /// With `enrich`, the returned page carries manifests and download totals
-    /// ([`Self::enrich_hits`]); a group searches its members without and
-    /// enriches only its own final page.
+    /// ([`Self::enrich_hits`]). A group reads the two halves apart: the ids
+    /// ([`Self::search_names`]), then the hits of its own page
+    /// ([`Self::hits_of`]), which it enriches.
     async fn search_filtered(
         &self,
         repo: &str,
@@ -103,10 +104,32 @@ impl HostedRepo {
         enrich: bool,
     ) -> Result<SearchResult, RepoError> {
         let repo_id = self.repo_id(repo).await?;
+        // Page over ids, not versions: versions are loaded for the returned
+        // page only.
+        let names = self.search_names(repo_id, params, filter).await?;
+        let total_hits = names.len();
+        let page: Vec<String> = names
+            .into_iter()
+            .skip(params.skip.min(total_hits))
+            .take(params.effective_take())
+            .collect();
+        let mut hits = self.hits_of(repo_id, params, &page).await?;
+        if enrich {
+            self.enrich_hits(repo_id, &mut hits).await?;
+        }
+        Ok(SearchResult { total_hits, hits })
+    }
+
+    /// The ids (stored, lowercase) of every package a search for `params`
+    /// finds, in byte order. The filters run in SQL, the selector and the
+    /// query's terms on the (much shorter) id list.
+    async fn search_names(
+        &self,
+        repo_id: i64,
+        params: &SearchParams,
+        filter: &PackageFilter,
+    ) -> Result<Vec<String>, RepoError> {
         let query = SearchQuery::parse(&params.q);
-        // Page over ids, not versions: the filters run in SQL, the selector
-        // and the query's terms on the (much shorter) id list, and versions
-        // are loaded for the returned page only.
         let mut names: Vec<String> = component_dao::search_names(
             &self.pool,
             repo_id,
@@ -121,19 +144,24 @@ impl HostedRepo {
         // Byte order, as group results are ordered (SQL orders by the
         // database collation, which may weigh punctuation differently).
         names.sort_unstable();
-        let total_hits = names.len();
-        let page: Vec<String> = names
-            .into_iter()
-            .skip(params.skip.min(total_hits))
-            .take(params.effective_take())
-            .collect();
+        Ok(names)
+    }
 
+    /// The unenriched search hits of `page` (ids [`Self::search_names`]
+    /// returned for `params`), in its order: each id's matching versions and
+    /// its casing. An id left without a matching version is skipped.
+    async fn hits_of(
+        &self,
+        repo_id: i64,
+        params: &SearchParams,
+        page: &[String],
+    ) -> Result<Vec<SearchHit>, RepoError> {
         use std::collections::HashMap;
         let mut by_id: HashMap<String, Vec<forge_db::model::Component>> = HashMap::new();
         for c in component_dao::search_versions(
             &self.pool,
             repo_id,
-            &page,
+            page,
             params.prerelease,
             params.include_semver2,
         )
@@ -142,7 +170,7 @@ impl HostedRepo {
             by_id.entry(c.name.clone()).or_default().push(c);
         }
         let mut hits: Vec<SearchHit> = Vec::with_capacity(page.len());
-        for name in &page {
+        for name in page {
             let Some(comps) = by_id.remove(name) else {
                 // Changed between the two queries.
                 continue;
@@ -165,10 +193,7 @@ impl HostedRepo {
                 ..Default::default()
             });
         }
-        if enrich {
-            self.enrich_hits(repo_id, &mut hits).await?;
-        }
-        Ok(SearchResult { total_hits, hits })
+        Ok(hits)
     }
 
     /// Attach the latest version's manifest and the download total to each
@@ -531,6 +556,29 @@ impl RepoReadOps for HostedRepo {
         let repo_id = self.repo_id(repo).await?;
         self.enrich_hits(repo_id, hits).await
     }
+
+    async fn search_ids(
+        &self,
+        repo: &str,
+        params: &SearchParams,
+        filter: &PackageFilter,
+    ) -> Result<Vec<String>, RepoError> {
+        let repo_id = self.repo_id(repo).await?;
+        self.search_names(repo_id, params, filter).await
+    }
+
+    async fn search_hits(
+        &self,
+        repo: &str,
+        params: &SearchParams,
+        ids: &[String],
+    ) -> Result<Vec<SearchHit>, RepoError> {
+        if ids.is_empty() {
+            return Ok(Vec::new());
+        }
+        let repo_id = self.repo_id(repo).await?;
+        self.hits_of(repo_id, params, ids).await
+    }
 }
 
 /// Storage-backed hosted **write** facade: publish, redeploy, delete, relist.

crates/core/src/repo/mod.rs

@@ -697,10 +697,10 @@ pub trait RepoReadOps: Send + Sync {
     /// the hits carry ids and versions, but [`SearchHit::nuspec_bytes`] and
     /// [`SearchHit::total_downloads`] may be left unset.
     ///
-    /// A group asks its local members for up to [`SearchParams::MAX_TAKE`]
-    /// hits, and a proxy that searches its upstream for the leading window of
-    /// its hits the client's page needs; it merges and counts them, and
-    /// renders only the client's page (see [`group`]).
+    /// A group asks a proxy that searches its upstream for the leading window
+    /// of its hits the client's page needs (its local members for their ids,
+    /// [`Self::search_ids`]); it merges and counts them, and renders only the
+    /// client's page (see [`group`]).
     /// It therefore searches its members through this method and enriches just
     /// the final page via [`Self::enrich_search_hits`], so manifest reads (S3)
     /// are bounded by the client's `take` rather than by every member's corpus.
@@ -727,6 +727,84 @@ pub trait RepoReadOps: Send + Sync {
     ) -> Result<(), RepoError> {
         Ok(())
     }
+
+    /// The id of every hit of a search of what `repo` holds itself: the hits
+    /// [`Self::search_unenriched`] finds for `params` without the upstream
+    /// (whatever `params.include_upstream`, `skip` and `take` say), in its
+    /// order. The hosted and proxy backends return the stored (lowercase)
+    /// ids, in byte order.
+    ///
+    /// A group pages its local members' ids rather than their hits, then
+    /// loads the hits of its page alone, each through the member that won it
+    /// ([`Self::search_hits`]): the versions of one page, not of every
+    /// member's leading hits (see [`group`]).
+    ///
+    /// The default pages through [`Self::search_unenriched`]: correct for any
+    /// implementation, merely not cheaper.
+    async fn search_ids(
+        &self,
+        repo: &str,
+        params: &SearchParams,
+        filter: &PackageFilter,
+    ) -> Result<Vec<String>, RepoError> {
+        Ok(all_local_hits(self, repo, params, filter)
+            .await?
+            .into_iter()
+            .map(|hit| hit.original_id)
+            .collect())
+    }
+
+    /// The unenriched hits ([`Self::search_unenriched`]) of `ids`, which
+    /// [`Self::search_ids`] returned for the same `params`, in the order of
+    /// `ids`. An id that is no hit any more (its last matching version was
+    /// deleted since) is left out.
+    ///
+    /// The default searches again ([`Self::search_ids`]' default).
+    async fn search_hits(
+        &self,
+        repo: &str,
+        params: &SearchParams,
+        ids: &[String],
+    ) -> Result<Vec<SearchHit>, RepoError> {
+        let mut hits: std::collections::HashMap<String, SearchHit> =
+            all_local_hits(self, repo, params, &PackageFilter::AllowAll)
+                .await?
+                .into_iter()
+                .map(|hit| (hit.original_id.clone(), hit))
+                .collect();
+        Ok(ids.iter().filter_map(|id| hits.remove(id)).collect())
+    }
+}
+
+/// Every unenriched hit of a search of what `repo` holds itself, page by
+/// page: the defaults of [`RepoReadOps::search_ids`] and
+/// [`RepoReadOps::search_hits`].
+async fn all_local_hits<R: RepoReadOps + ?Sized>(
+    ops: &R,
+    repo: &str,
+    params: &SearchParams,
+    filter: &PackageFilter,
+) -> Result<Vec<SearchHit>, RepoError> {
+    let mut hits = Vec::new();
+    loop {
+        let page = ops
+            .search_unenriched(
+                repo,
+                &SearchParams {
+                    skip: hits.len(),
+                    take: SearchParams::MAX_TAKE,
+                    include_upstream: false,
+                    ..params.clone()
+                },
+                filter,
+            )
+            .await?;
+        let returned = page.hits.len();
+        hits.extend(page.hits);
+        if returned == 0 || hits.len() >= page.total_hits {
+            return Ok(hits);
+        }
+    }
 }
 
 /// A package-filtering decorator around any repository read implementation.
@@ -899,6 +977,28 @@ impl RepoReadOps for PackageFilteredReadOps {
         // the predicate; enrichment adds no package identities.
         self.inner.enrich_search_hits(repo, hits).await
     }
+
+    async fn search_ids(
+        &self,
+        repo: &str,
+        params: &SearchParams,
+        filter: &PackageFilter,
+    ) -> Result<Vec<String>, RepoError> {
+        let combined = self.filter.clone().intersect(filter.clone());
+        self.inner.search_ids(repo, params, &combined).await
+    }
+
+    async fn search_hits(
+        &self,
+        repo: &str,
+        params: &SearchParams,
+        ids: &[String],
+    ) -> Result<Vec<SearchHit>, RepoError> {
+        // The ids came out of the filtered search above; an id named any other
+        // way is refused all the same.
+        let ids: Vec<String> = ids.iter().filter(|id| self.allows(id)).cloned().collect();
+        self.inner.search_hits(repo, params, &ids).await
+    }
 }
 
 /// A boxed async byte source, used by [`RepoWriteOps::store_upload`] so the