Merge pull request #258 from holaplex/espi/count-caching

Redis Cache Total Mints and Supply

api/src/dataloaders/collection.rs

@@ -2,9 +2,17 @@ use std::collections::HashMap;
 
 use async_graphql::{dataloader::Loader as DataLoader, FieldError, Result};
 use poem::async_trait;
-use sea_orm::prelude::*;
+use redis::{AsyncCommands, Client as Redis};
+use sea_orm::{prelude::*, FromQueryResult, QueryFilter, QuerySelect};
 
-use crate::{db::Connection, entities::collections, objects::Collection};
+use crate::{
+    db::Connection,
+    entities::{
+        collection_mints, collections, drops,
+        sea_orm_active_enums::{CreationStatus, DropType},
+    },
+    objects::Collection,
+};
 
 #[derive(Debug, Clone)]
 pub struct Loader {
@@ -35,3 +43,188 @@ impl DataLoader<Uuid> for Loader {
             .collect()
     }
 }
+
+#[derive(FromQueryResult, Debug, Clone)]
+struct CollectionTotalMintsCount {
+    id: Uuid,
+    count: i64,
+}
+
+#[derive(Debug, Clone)]
+pub struct TotalMintsLoader {
+    pub db: Connection,
+    pub redis: Redis,
+}
+
+impl TotalMintsLoader {
+    #[must_use]
+    pub fn new(db: Connection, redis: Redis) -> Self {
+        Self { db, redis }
+    }
+}
+
+#[async_trait]
+impl DataLoader<Uuid> for TotalMintsLoader {
+    type Error = FieldError;
+    type Value = i64;
+
+    async fn load(&self, keys: &[Uuid]) -> Result<HashMap<Uuid, Self::Value>, Self::Error> {
+        let mut results: HashMap<Uuid, Self::Value> = HashMap::new();
+        let mut missing_keys: Vec<Uuid> = Vec::new();
+
+        let mut redis_connection = self.redis.get_async_connection().await?;
+
+        for key in keys {
+            let redis_key = format!("collection:{key}:total_mints");
+            match redis_connection.get::<_, i64>(&redis_key).await {
+                Ok(value) => {
+                    results.insert(*key, value);
+                },
+                Err(_) => {
+                    missing_keys.push(*key);
+                },
+            }
+        }
+
+        if missing_keys.is_empty() {
+            return Ok(results);
+        }
+
+        let conn = self.db.get();
+        let count_results = collection_mints::Entity::find()
+            .select_only()
+            .column_as(collection_mints::Column::Id.count(), "count")
+            .column_as(collection_mints::Column::CollectionId, "id")
+            .filter(
+                collection_mints::Column::CollectionId
+                    .is_in(missing_keys.iter().map(ToOwned::to_owned))
+                    .and(collection_mints::Column::CreationStatus.ne(CreationStatus::Queued)),
+            )
+            .group_by(collection_mints::Column::CollectionId)
+            .into_model::<CollectionTotalMintsCount>()
+            .all(conn)
+            .await?;
+        let count_results = count_results
+            .into_iter()
+            .map(|result| (result.id, result.count))
+            .collect::<HashMap<_, _>>();
+
+        for key in missing_keys {
+            let count = count_results.get(&key).copied().unwrap_or_default();
+            let redis_key = format!("collection:{key}:total_mints");
+
+            redis_connection
+                .set::<_, i64, Option<i64>>(&redis_key, count)
+                .await?;
+
+            results.insert(key, count);
+        }
+
+        Ok(results)
+    }
+}
+
+#[derive(FromQueryResult)]
+struct CollectionSupplyCount {
+    id: Uuid,
+    count: i64,
+}
+
+#[derive(Debug, Clone)]
+pub struct SupplyLoader {
+    pub db: Connection,
+    pub redis: Redis,
+}
+
+impl SupplyLoader {
+    #[must_use]
+    pub fn new(db: Connection, redis: Redis) -> Self {
+        Self { db, redis }
+    }
+}
+
+#[async_trait]
+impl DataLoader<Uuid> for SupplyLoader {
+    type Error = FieldError;
+    type Value = Option<i64>;
+
+    async fn load(&self, keys: &[Uuid]) -> Result<HashMap<Uuid, Self::Value>, Self::Error> {
+        let mut results: HashMap<Uuid, Self::Value> = HashMap::new();
+        let mut missing_keys: Vec<Uuid> = Vec::new();
+
+        let mut redis_connection = self.redis.get_async_connection().await?;
+
+        for key in keys {
+            let redis_key = format!("collection:{key}:supply");
+            match redis_connection.get::<_, Option<i64>>(&redis_key).await {
+                Ok(value) => {
+                    results.insert(*key, value);
+                },
+                Err(_) => {
+                    missing_keys.push(*key);
+                },
+            }
+        }
+
+        if missing_keys.is_empty() {
+            return Ok(results);
+        }
+
+        let conn = self.db.get();
+        let mut computed_supplies: Vec<Uuid> = Vec::new();
+
+        let collection_with_drops = collections::Entity::find()
+            .filter(collections::Column::Id.is_in(missing_keys.iter().map(ToOwned::to_owned)))
+            .inner_join(drops::Entity)
+            .select_also(drops::Entity)
+            .all(conn)
+            .await?;
+
+        for (collection, drop) in collection_with_drops {
+            if let Some(drop) = drop {
+                if drop.drop_type == DropType::Open {
+                    computed_supplies.push(collection.id);
+                    continue;
+                }
+                continue;
+            }
+
+            let redis_key = format!("collection:{}:supply", collection.id);
+
+            let supply = redis_connection
+                .set::<_, Option<i64>, Option<i64>>(&redis_key, collection.supply)
+                .await?;
+
+            results.insert(collection.id, supply);
+        }
+
+        let count_results = collection_mints::Entity::find()
+            .select_only()
+            .column_as(collection_mints::Column::Id.count(), "count")
+            .column_as(collection_mints::Column::CollectionId, "id")
+            .filter(
+                collection_mints::Column::CollectionId
+                    .is_in(computed_supplies.iter().map(ToOwned::to_owned)),
+            )
+            .group_by(collection_mints::Column::CollectionId)
+            .into_model::<CollectionSupplyCount>()
+            .all(conn)
+            .await?
+            .into_iter()
+            .map(|result| (result.id, result.count))
+            .collect::<HashMap<_, _>>();
+
+        for key in computed_supplies {
+            let count = count_results.get(&key).copied().unwrap_or_default();
+            let redis_key = format!("collection:{key}:supply");
+
+            let count = redis_connection
+                .set::<_, Option<i64>, Option<i64>>(&redis_key, Some(count))
+                .await?;
+
+            results.insert(key, count);
+        }
+
+        Ok(results)
+    }
+}

api/src/dataloaders/collection_drop.rs

@@ -2,13 +2,9 @@ use std::collections::HashMap;
 
 use async_graphql::{dataloader::Loader as DataLoader, FieldError, Result};
 use poem::async_trait;
-use sea_orm::{prelude::*, JoinType, QuerySelect};
+use sea_orm::prelude::*;
 
-use crate::{
-    db::Connection,
-    entities::{collections, drops},
-    objects::Drop,
-};
+use crate::{db::Connection, entities::drops, objects::Drop};
 
 #[derive(Debug, Clone)]
 pub struct Loader {
@@ -29,26 +25,13 @@ impl DataLoader<Uuid> for Loader {
 
     async fn load(&self, keys: &[Uuid]) -> Result<HashMap<Uuid, Self::Value>, Self::Error> {
         let drops = drops::Entity::find()
-            .join(JoinType::InnerJoin, drops::Relation::Collections.def())
-            .select_also(collections::Entity)
             .filter(drops::Column::CollectionId.is_in(keys.iter().map(ToOwned::to_owned)))
             .all(self.db.get())
             .await?;
 
-        drops
+        Ok(drops
             .into_iter()
-            .map(|(drop, collection)| {
-                Ok((
-                    drop.collection_id,
-                    Drop::new(
-                        drop.clone(),
-                        collection.ok_or(FieldError::new(format!(
-                            "no collection for the drop {}",
-                            drop.id
-                        )))?,
-                    ),
-                ))
-            })
-            .collect::<Result<HashMap<Uuid, Self::Value>>>()
+            .map(|drop| (drop.collection_id, drop.into()))
+            .collect::<HashMap<_, _>>())
     }
 }

api/src/dataloaders/drop.rs

@@ -1,53 +1,37 @@
 use std::collections::HashMap;
 
-use async_graphql::{dataloader::Loader as DataLoader, FieldError, Result};
+use async_graphql::{dataloader::Loader, FieldError, Result};
 use poem::async_trait;
-use sea_orm::{prelude::*, JoinType, QuerySelect};
+use sea_orm::prelude::*;
 
-use crate::{
-    db::Connection,
-    entities::{collections, drops},
-    objects::Drop,
-};
+use crate::{db::Connection, entities::drops, objects::Drop};
 
 #[derive(Debug, Clone)]
-pub struct Loader {
+pub struct DropLoader {
     pub db: Connection,
 }
 
-impl Loader {
+impl DropLoader {
     #[must_use]
     pub fn new(db: Connection) -> Self {
         Self { db }
     }
 }
 
 #[async_trait]
-impl DataLoader<Uuid> for Loader {
+impl Loader<Uuid> for DropLoader {
     type Error = FieldError;
     type Value = Drop;
 
     async fn load(&self, keys: &[Uuid]) -> Result<HashMap<Uuid, Self::Value>, Self::Error> {
         let drops = drops::Entity::find()
-            .join(JoinType::InnerJoin, drops::Relation::Collections.def())
-            .select_also(collections::Entity)
             .filter(drops::Column::Id.is_in(keys.iter().map(ToOwned::to_owned)))
             .all(self.db.get())
             .await?;
 
-        drops
+        Ok(drops
             .into_iter()
-            .map(|(drop, collection)| {
-                Ok((
-                    drop.id,
-                    Drop::new(
-                        drop.clone(),
-                        collection.ok_or_else(|| {
-                            FieldError::new(format!("no collection for the drop {}", drop.id))
-                        })?,
-                    ),
-                ))
-            })
-            .collect::<Result<HashMap<Uuid, Self::Value>>>()
+            .map(|drop| (drop.id, drop.into()))
+            .collect())
     }
 }

api/src/dataloaders/drops.rs

@@ -2,13 +2,9 @@ use std::collections::HashMap;
 
 use async_graphql::{dataloader::Loader as DataLoader, FieldError, Result};
 use poem::async_trait;
-use sea_orm::{prelude::*, JoinType, QuerySelect};
+use sea_orm::prelude::*;
 
-use crate::{
-    db::Connection,
-    entities::{collections, drops},
-    objects::Drop,
-};
+use crate::{db::Connection, entities::drops, objects::Drop};
 
 #[derive(Debug, Clone)]
 pub struct ProjectLoader {
@@ -29,17 +25,13 @@ impl DataLoader<Uuid> for ProjectLoader {
 
     async fn load(&self, keys: &[Uuid]) -> Result<HashMap<Uuid, Self::Value>, Self::Error> {
         let drops = drops::Entity::find()
-            .join(JoinType::InnerJoin, drops::Relation::Collections.def())
-            .select_also(collections::Entity)
             .filter(drops::Column::ProjectId.is_in(keys.iter().map(ToOwned::to_owned)))
             .all(self.db.get())
             .await?;
 
         Ok(drops
             .into_iter()
-            .filter_map(|(drop, collection)| {
-                collection.map(|collection| (drop.project_id, Drop::new(drop, collection)))
-            })
+            .map(|drop| (drop.project_id, drop.into()))
             .fold(HashMap::new(), |mut acc, (project, drop)| {
                 acc.entry(project).or_insert_with(Vec::new);
 

api/src/dataloaders/mod.rs

@@ -14,14 +14,17 @@ mod project_collections;
 mod switch_collection_histories;
 mod update_histories;
 
-pub use collection::Loader as CollectionLoader;
+pub use collection::{
+    Loader as CollectionLoader, SupplyLoader as CollectionSupplyLoader,
+    TotalMintsLoader as CollectionTotalMintsLoader,
+};
 pub use collection_drop::Loader as CollectionDropLoader;
 pub use collection_mints::{
     CollectionMintLoader, Loader as CollectionMintsLoader,
     OwnerLoader as CollectionMintsOwnerLoader, QueuedMintsLoader,
 };
 pub use creators::Loader as CreatorsLoader;
-pub use drop::Loader as DropLoader;
+pub use drop::DropLoader;
 pub use drops::ProjectLoader as ProjectDropsLoader;
 pub use holders::Loader as HoldersLoader;
 pub use metadata_json::{

api/src/entities/collection_mints.rs

@@ -88,4 +88,12 @@ impl Entity {
             .select_also(collections::Entity)
             .filter(Column::Id.eq(id))
     }
+
+    pub fn filter_by_collection(id: Uuid) -> Select<Self> {
+        Self::find().filter(
+            Column::CollectionId
+                .eq(id)
+                .and(Column::CreationStatus.ne(CreationStatus::Queued)),
+        )
+    }
 }

api/src/entities/collections.rs

@@ -15,7 +15,6 @@ pub struct Model {
     #[sea_orm(nullable)]
     pub credits_deduction_id: Option<Uuid>,
     pub creation_status: CreationStatus,
-    pub total_mints: i64,
     #[sea_orm(column_type = "Text", nullable)]
     pub address: Option<String>,
     #[sea_orm(nullable)]