Using proper counter for filtered documents

integrationos-api/src/endpoints/mod.rs

@@ -11,7 +11,6 @@ use axum::{
     Extension, Json,
 };
 use bson::{doc, SerializerOptions};
-use futures::TryFutureExt;
 use http::{HeaderMap, HeaderValue, StatusCode};
 use integrationos_cache::local::connection_cache::ConnectionCacheArcStrHeaderKey;
 use integrationos_domain::{
@@ -22,7 +21,6 @@ use mongodb::options::FindOneOptions;
 use serde::{de::DeserializeOwned, Deserialize, Serialize};
 use serde_json::Value;
 use std::{collections::BTreeMap, fmt::Debug, sync::Arc};
-use tokio::try_join;
 use tracing::error;
 
 pub mod common_enum;
@@ -185,30 +183,20 @@ where
     );
 
     let store = T::get_store(state.app_stores.clone());
-    let total = store
-        .collection
-        .estimated_document_count(None)
-        .map_err(|e| {
-            error!("Error counting documents: {e}");
-            internal_server_error!()
-        })
-        .await?;
-    let find = store.get_many(
+    let find = store.get_many_with_count(
         Some(query.filter),
         None,
-        Some(doc! {
-            "createdAt": -1
-        }),
+        None,
         Some(query.limit),
         Some(query.skip),
     );
 
-    let res = match try_join!(find) {
-        Ok((rows,)) => ReadResponse {
+    let res = match find.await {
+        Ok((rows, total)) => ReadResponse {
             rows: rows.into_iter().map(T::public).collect(),
             skip: query.skip,
             limit: query.limit,
-            total,
+            total: total as u64,
         },
         Err(e) => {
             error!("Error reading from store: {e}");

integrationos-domain/src/algebra/store.rs

@@ -1,6 +1,7 @@
 use crate::IntegrationOSError;
 use crate::Store;
 use bson::doc;
+use futures::StreamExt;
 use futures::TryStreamExt;
 use mongodb::bson::Document;
 use mongodb::options::CountOptions;
@@ -56,6 +57,19 @@ impl<T: Serialize + DeserializeOwned + Unpin + Sync + Send + 'static> MongoStore
         limit: Option<u64>,
         skip: Option<u64>,
     ) -> Result<Vec<T>, IntegrationOSError> {
+        self.get_many_with_count(filter, selection, sort, limit, skip)
+            .await
+            .map(|(r, _)| r)
+    }
+
+    pub async fn get_many_with_count(
+        &self,
+        filter: Option<Document>,
+        selection: Option<Document>,
+        sort: Option<Document>,
+        limit: Option<u64>,
+        skip: Option<u64>,
+    ) -> Result<(Vec<T>, usize), IntegrationOSError> {
         let mut filter_options = mongodb::options::FindOptions::default();
         filter_options.sort = sort;
         filter_options.projection = selection;
@@ -66,10 +80,11 @@ impl<T: Serialize + DeserializeOwned + Unpin + Sync + Send + 'static> MongoStore
             filter_options.sort = Some(doc! { "createdAt": -1 });
         }
 
-        let cursor = self.collection.find(filter, filter_options).await?;
+        let mut cursor = self.collection.find(filter.clone(), filter_options).await?;
+        let count = cursor.by_ref().count().await;
         let records = cursor.try_collect().await?;
 
-        Ok(records)
+        Ok((records, count.to_owned()))
     }
 
     pub async fn create_one(&self, data: &T) -> Result<(), IntegrationOSError> {