Upgrade discovery to use meilisearch

Author: lucemans

Cargo.lock

@@ -75,4 +75,3 @@ prometheus = "0.13.4"
 time = "0.3.36"
 dashmap = "6.1.0"
 meilisearch-sdk = "0.28.0"
-clickhouse = { version = "0.13.1", features = ["rustls-tls"] }

packages/app/Cargo.toml

@@ -42,7 +42,8 @@ services:
     ports:
       - 7700:7700
     environment:
-      MEILISEARCH_NO_ANALYTICS: true
+      MEILI_NO_ANALYTICS: true
+      MEILI_MASTER_KEY: admin
   clickhouse:
     image: clickhouse
     ports:

packages/app/compose.yaml

@@ -1,77 +1,49 @@
+use std::collections::HashMap;
+
 use axum::async_trait;
 use enstate_shared::core::Profile;
 use enstate_shared::discovery::Discovery;
+use ethers::providers::namehash;
+use serde::{Deserialize, Serialize};
 
 pub struct DiscoveryEngine {
-    client: clickhouse::Client,
+    client: meilisearch_sdk::client::Client,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct MeiliProfileDocument {
+    name_hash: String,
+    name: String,
+    avatar: Option<String>,
+    header: Option<String>,
+    display: String,
+    addresses: HashMap<String, String>,
+    fresh: i64,
+}
+
+impl From<&Profile> for MeiliProfileDocument {
+    fn from(profile: &Profile) -> Self {
+        Self {
+            name_hash: format!("{:x}", namehash(&profile.name)),
+            name: profile.name.clone(),
+            avatar: profile.avatar.clone(),
+            header: profile.header.clone(),
+            display: profile.display.clone(),
+            addresses: profile.chains.iter().map(|(chain, address)| (chain.to_string(), address.to_string())).collect(),
+            fresh: profile.fresh,
+        }
+    }
 }
 
 impl DiscoveryEngine {
-    pub fn new(url: &str, user: &str, password: &str) -> Self {
+    pub fn new(url: &str, key: Option<&str>) -> Self {
         Self {
-            client: clickhouse::Client::default().with_url(url).with_user(user).with_password(password),
+            client: meilisearch_sdk::client::Client::new(url, key).unwrap(),
         }
     }
 
     pub async fn create_table_if_not_exists(&self) -> Result<(), ()> {
-        // Create the profiles table
-        let profiles_table_query = self.client.query("CREATE TABLE IF NOT EXISTS enstate.profiles (name String, address String, avatar String, header String, display String, contenthash String, resolver String, ccip_urls Array(String), errors Array(String), fresh Int64) ENGINE = ReplacingMergeTree(fresh) ORDER BY name");
-
-        match profiles_table_query.execute().await {
-            Ok(_) => {},
-            Err(e) => {
-                tracing::error!("Error creating profiles table: {}", e); 
-                return Err(());
-            }
-        };
 
-        // Create the stats aggregation table
-        let stats_table_query = self.client.query("
-            CREATE TABLE IF NOT EXISTS enstate.profile_stats_agg
-            (
-                day                 Date,
-                total_profiles      UInt64,
-                profiles_with_avatar UInt64,
-                profiles_with_header UInt64,
-                profiles_with_display UInt64,
-                profiles_with_contenthash UInt64
-            )
-            ENGINE = SummingMergeTree
-            PARTITION BY day
-            ORDER BY day
-        ");
-        
-        match stats_table_query.execute().await {
-            Ok(_) => {},
-            Err(e) => {
-                tracing::error!("Error creating profile_stats_agg table: {}", e); 
-                return Err(());
-            }
-        };
-        
-        // Create the materialized view
-        let materialized_view_query = self.client.query("
-            CREATE MATERIALIZED VIEW IF NOT EXISTS enstate.mv_profile_stats_agg
-            TO enstate.profile_stats_agg
-            AS
-            SELECT
-                toDate(fromUnixTimestamp64Milli(fresh)) AS day,
-                count() AS total_profiles,
-                countIf(avatar != '') AS profiles_with_avatar,
-                countIf(header != '') AS profiles_with_header,
-                countIf(display != '') AS profiles_with_display,
-                countIf(contenthash != '') AS profiles_with_contenthash
-            FROM enstate.profiles
-            GROUP BY day
-        ");
-        
-        match materialized_view_query.execute().await {
-            Ok(_) => {},
-            Err(e) => {
-                tracing::error!("Error creating materialized view: {}", e); 
-                return Err(());
-            }
-        };
 
         Ok(())
     }
@@ -88,21 +60,31 @@ impl Discovery for DiscoveryEngine {
             .map(|(key, val)| format!("{}: {}", key, val))
             .collect();
 
-        let query = self.client
-            .query("INSERT INTO enstate.profiles (name, address, avatar, header, display, contenthash, resolver, ccip_urls, errors, fresh) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)")
-            .bind(profile.name.clone())
-            .bind(profile.address.as_ref().map(|x| x.to_string()).unwrap_or_default())
-            .bind(profile.avatar.as_deref().unwrap_or_default())
-            .bind(profile.header.as_deref().unwrap_or_default())
-            .bind(profile.display.clone())
-            .bind(profile.contenthash.as_deref().unwrap_or_default())
-            .bind(profile.resolver.clone())
-            .bind(ccip_urls)
-            .bind(errors)
-            .bind(profile.fresh);
+        // let query = self.client
+        //     .query("INSERT INTO enstate.profiles (name, address, avatar, header, display, contenthash, resolver, ccip_urls, errors, fresh) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)")
+        //     .bind(profile.name.clone())
+        //     .bind(profile.address.as_ref().map(|x| x.to_string()).unwrap_or_default())
+        //     .bind(profile.avatar.as_deref().unwrap_or_default())
+        //     .bind(profile.header.as_deref().unwrap_or_default())
+        //     .bind(profile.display.clone())
+        //     .bind(profile.contenthash.as_deref().unwrap_or_default())
+        //     .bind(profile.resolver.clone())
+        //     .bind(ccip_urls)
+        //     .bind(errors)
+        //     .bind(profile.fresh);
+
+        let document = MeiliProfileDocument::from(profile);
+
+        let documents = vec![document];
+        
+        let x = self.client.index("profiles");
+        let x = x.add_documents(&documents, Some("name_hash")).await;
 
-        match query.execute().await {
-            Ok(_) => Ok(()),
+        match x {
+            Ok(result) => {
+                tracing::info!("Inserted profile: {:?}", result);
+                Ok(())
+            },
             Err(e) => {
                 tracing::error!("Error inserting profile: {}", e);
                 Err(())

packages/app/src/discovery/engine.rs

@@ -3,6 +3,7 @@ use std::sync::Arc;
 
 use enstate_shared::cache::{CacheLayer, PassthroughCacheLayer};
 use enstate_shared::core::ENSService;
+use enstate_shared::discovery::Discovery;
 use enstate_shared::models::{
     multicoin::cointype::{coins::CoinType, Coins},
     records::Records,
@@ -90,14 +91,19 @@ impl AppState {
         let cache_ttl =
             env::var("PROFILE_CACHE_TTL").map_or(Some(600), |cache_ttl| cache_ttl.parse().ok());
 
-        let discovery_engine = DiscoveryEngine::new("http://localhost:8123", "admin", "admin");
+        let meilisearch_url = env::var("MEILI_ENDPOINT").ok();
+        let meilisearch_key = env::var("MEILI_KEY").ok();
 
-        discovery_engine.create_table_if_not_exists().await;
+        let discovery = meilisearch_url.map(|url| {
+            let engine = DiscoveryEngine::new(&url, meilisearch_key.as_deref());
+
+            Box::new(engine) as Box<dyn Discovery>
+        });
 
         Self {
             rate_limiter: RateLimiter::new(),
             service: ENSService {
-                discovery: Some(Box::new(discovery_engine)),
+                discovery,
                 cache,
                 rpc: Box::new(provider),
                 opensea_api_key,