feat(migration): skip Cs+ profile if doesn't exist on squid

.env.example

@@ -5,6 +5,7 @@ HASURA_GRAPHQL_ADMIN_SECRET=my_hasura_password
 HASURA_LISTEN_PORT=8080
 DUNITER_ENDPOINTS="['wss://gdev.p2p.legal/ws', 'wss://gdev.coinduf.eu/ws', 'wss://gdev.cgeek.fr/ws']"
 
-### OPTIONNAL ###
+### OPTIONNAL: only for Cs+ data migration ###
 IMPORT_CSPLUS_DATA=false
 CSPLUS_ENDPOINT="https://g1.data.e-is.pro"
+SQUID_ENDPOINT="https://subsquid.gdev.coinduf.eu/graphql"

docker-compose.prod.yml

@@ -15,5 +15,6 @@ services:
       DB_DATABASE: ${DB_DATABASE}
       IMPORT_CSPLUS_DATA: ${IMPORT_CSPLUS_DATA}
       DUNITER_ENDPOINTS: ${DUNITER_ENDPOINTS}
-      CSPLUS_ENDPOINT: ${CSPLUS_ENDPOINT}
       HASURA_LISTEN_PORT: ${HASURA_LISTEN_PORT}
+      CSPLUS_ENDPOINT: ${CSPLUS_ENDPOINT}
+      SQUID_ENDPOINT: ${SQUID_ENDPOINT}

lib/utils.ts

@@ -50,8 +50,6 @@ export async function runCsplusImport(isProduction: boolean) {
   if (!status.success) {
     throw new Error(`Process migrate_csplus exited with code: ${status.code}`);
   }
-
-  console.log("End of Cs+ data import");
 }
 
 export async function isProfilesTableEmpty(client: Client): Promise<boolean> {

migrate_csplus/src/data_fetch.rs

@@ -7,6 +7,7 @@ use tokio_postgres::NoTls;
 
 use crate::{
     db::insert_profiles,
+    fetch_addresses_from_squid::fetch_ids_from_graphql,
     models::{Hit, ScrollResponse},
 };
 
@@ -27,16 +28,22 @@ pub async fn fetch_profiles(
     let total_profiles = fetch_total_profiles_count(&url_profiles).await?;
     let total_pages = (total_profiles as f64 / page_size as f64).ceil() as usize;
 
-    println!("Start migration of {} CS+ profiles...", total_profiles);
+    println!(
+        "Start migration of {} CS+ profiles from {} endpoint ...",
+        total_profiles, csplus_endpoint
+    );
     println!("Retrieving page: {}/{}", iteration, total_pages);
 
     // Fetch initial batch
     let mut future_scroll_response =
         fetch_batch(url_profiles.to_string(), initial_batch_body).await;
 
+    // Fetch IDs from the GraphQL endpoint
+    let ids = Arc::new(fetch_ids_from_graphql().await?);
+
     while let Ok(current_response) = future_scroll_response {
         // Process the current batch
-        process_hits(pool_arc.clone(), &current_response.hits.hits).await?;
+        process_hits(pool_arc.clone(), &current_response.hits.hits, &ids).await?;
 
         // Log the retrieval or completion
         iteration += 1;
@@ -77,6 +84,7 @@ async fn fetch_batch(
 async fn process_hits(
     pool: Arc<Mutex<Pool<PostgresConnectionManager<NoTls>>>>,
     hits: &[Hit],
+    ids: &Arc<Vec<String>>,
 ) -> Result<(), Box<dyn Error>> {
     println!("Processing...");
     let mut tasks = Vec::new();
@@ -84,10 +92,12 @@ async fn process_hits(
     for chunk in hits.chunks(2000) {
         let pool_clone = pool.clone();
         let chunk_clone = chunk.to_vec();
+        let ids_clone = Arc::clone(&ids);
 
         let task = tokio::spawn(async move {
             if let Ok(mut client) = pool_clone.lock().await.get().await {
-                if let Err(e) = insert_profiles(&mut client, &chunk_clone).await {
+                if let Err(e) = insert_profiles(&mut client, &chunk_clone, ids_clone.as_ref()).await
+                {
                     eprintln!("Insert error: {}", e);
                 }
             }

migrate_csplus/src/db.rs

@@ -42,12 +42,25 @@ pub async fn database_connection(
 }
 
 /// Insert parsed profiles to Postgres database
-pub async fn insert_profiles(client: &mut Client, hits: &[Hit]) -> Result<(), Box<dyn Error>> {
+pub async fn insert_profiles(
+    client: &mut Client,
+    hits: &[Hit],
+    ids: &[String],
+) -> Result<(), Box<dyn Error>> {
     let transaction = client.transaction().await?;
 
     for hit in hits {
         let profile_row = ProfileRow::from_hit(hit)?;
 
+        // Skip insert if address is not in the list of IDs
+        if profile_row
+            .address
+            .as_ref()
+            .map_or(true, |address| !ids.contains(address))
+        {
+            continue;
+        }
+
         // Construct the query using optional parameters for nullable fields
         let statement = transaction.prepare(
             "INSERT INTO public.profiles (address, avatar, description, geoloc, title, city, socials, created_at)

migrate_csplus/src/fetch_addresses_from_squid.rs

+use reqwest;
+use serde::Deserialize;
+use serde_json::json;
+use std::{env, error::Error};
+
+// Define a structure to deserialize the GraphQL response
+#[derive(Deserialize)]
+struct GraphQLResponse {
+    data: GraphQLData,
+}
+
+#[derive(Deserialize)]
+struct GraphQLData {
+    accounts: Vec<Account>,
+}
+
+#[derive(Deserialize)]
+struct Account {
+    id: String,
+}
+
+/// Get all existing accounts on squid indexer
+pub async fn fetch_ids_from_graphql() -> Result<Vec<String>, Box<dyn Error>> {
+    let endpoint = env::var("SQUID_ENDPOINT").map_err(|_| "SQUID_ENDPOINT must be set")?;
+    println!("Fetch addresses from squid endpoint {} ...", endpoint);
+
+    // Define the GraphQL query
+    let graphql_query = json!({
+        "query": "{ accounts { id } }"
+    });
+
+    // Send the request
+    let client = reqwest::Client::new();
+    let res = client.post(endpoint).json(&graphql_query).send().await?;
+
+    // Check if the request was successful
+    if res.status().is_success() {
+        // Parse the response
+        let response_body: GraphQLResponse = res.json().await?;
+        // Extract the IDs
+        let ids = response_body
+            .data
+            .accounts
+            .into_iter()
+            .map(|account| account.id)
+            .collect();
+        Ok(ids)
+    } else {
+        Err(Box::new(std::io::Error::new(
+            std::io::ErrorKind::Other,
+            "Failed to fetch data from GraphQL endpoint",
+        )))
+    }
+}

migrate_csplus/src/main.rs

@@ -2,6 +2,7 @@ mod data_fetch;
 mod db;
 mod models;
 mod utils;
+mod fetch_addresses_from_squid;
 
 use crate::db::database_connection;