Merge pull request #17 from OpenArchive/feat/refresh-and-health-apis

Feat/refresh and health apis

Author: tripledoublev

Cargo.lock

@@ -44,9 +44,14 @@ num_cpus = "1.16.0"
 tokio-stream = "0.1.16"
 bytes = "1.7.2"
 iroh-blobs = "0.24.0"
+log = "0.4"
 
 [target.'cfg(target_os = "android")'.dependencies]
 jni = "0.21.1"
 tokio = { version = "~1.39",  default-features = false,  features = ["rt", "rt-multi-thread", "sync", "time", "macros"] }
 veilid-core = { git = "https://gitlab.com/veilid/veilid.git", version = "0.4.3" }
 blake3 = "1.5.4"
+
+[dev-dependencies]
+env_logger = "0.10"
+serial_test = "2.0"

Cargo.toml

@@ -15,6 +15,7 @@ use std::error::Error;
 use std::sync::{Arc, Mutex, Once};
 use std::thread;
 use veilid_core::veilid_core_setup_android;
+use std::time::Duration;
 
 
 trait IntoJObject {
@@ -104,8 +105,47 @@ pub extern "system" fn Java_net_opendasharchive_openarchive_services_snowbird_Sn
     _clazz: JClass,
     ctx: JObject,
 ) -> jstring {
+    log_debug!(TAG, "Bridge: stopping server");
+
+    // Create a runtime to handle async operations
+    let runtime = tokio::runtime::Runtime::new().unwrap();
+    
+    // Stop the backend server and clean up Veilid API
+    let stop_result = runtime.block_on(async {
+        // First stop the backend
+        match crate::server::server::stop().await {
+            Ok(_) => {
+                log_info!(TAG, "Backend stopped successfully");
+                
+                // Get the backend to access Veilid API
+                if let Ok(mut backend) = crate::server::server::get_backend().await {
+                    // Shutdown Veilid API
+                    if let Some(veilid_api) = backend.get_veilid_api() {
+                        veilid_api.shutdown().await;
+                        log_info!(TAG, "Veilid API shut down successfully");
+                    }
+                }
+                
+                // Add a small delay to ensure tasks complete
+                tokio::time::sleep(Duration::from_millis(500)).await;
+                
+                Ok(())
+            }
+            Err(e) => {
+                log_error!(TAG, "Error stopping server: {:?}", e);
+                Err(e)
+            }
+        }
+    });
+
+    // Create response string based on result
+    let response = match stop_result {
+        Ok(_) => "Server stopped successfully",
+        Err(_) => "Error stopping server",
+    };
+
     let output = env
-        .new_string("Server stopped")
+        .new_string(response)
         .expect("Couldn't create java string!");
 
     output.into_raw()

src/android_bridge.rs

@@ -120,47 +120,124 @@ async fn refresh_group(group_id: web::Path<String>) -> AppResult<impl Responder>
     let key = create_veilid_cryptokey_from_base64(group_id.as_str())?;
     log_debug!(TAG, "Got key {}", key);
 
-    let group = backend.get_group(&key).await?;
+    // Return error if group not found
+    let group = match backend.get_group(&key).await {
+        Ok(group) => group,
+        Err(e) => {
+            return Ok(HttpResponse::NotFound().json(json!({
+                "status": "error",
+                "error": format!("Group not found: {}", e)
+            })));
+        }
+    };
     log_debug!(TAG, "Got group");
 
     // Get all repos in the group
     let locked_repos = group.repos.lock().await;
     let repos: Vec<_> = locked_repos.values().cloned().collect();
     drop(locked_repos); // Release the lock before async operations
 
-    let mut refreshed_files = Vec::new();
+    // Return empty arrays if no repos
+    if repos.is_empty() {
+        return Ok(HttpResponse::Ok().json(json!({
+            "status": "success",
+            "refreshed_files": [],
+            "repos": []
+        })));
+    }
+
+    let mut refreshed_repos = Vec::new();
 
     // For each repo, refresh its collection and files
     for repo in repos {
         log_debug!(TAG, "Refreshing repo {}", repo.id());
-        
-        // Refresh collection hash
-        let collection_hash = repo.get_hash_from_dht().await?;
-        if !group.has_hash(&collection_hash).await? {
-            group.download_hash_from_peers(&collection_hash).await?;
-            log_debug!(TAG, "Downloaded collection hash {}", collection_hash);
-        }
 
-        // Get and refresh all files
-        let files = repo.list_files().await?;
-        for file_name in files {
-            match repo.get_file_hash(&file_name).await {
-                Ok(file_hash) => {
-                    if !group.has_hash(&file_hash).await? {
-                        group.download_hash_from_peers(&file_hash).await?;
-                        log_debug!(TAG, "Downloaded file hash {} for {}", file_hash, file_name);
-                        refreshed_files.push(file_name);
+        let mut repo_info = json!({
+            "repo_id": repo.id().to_string(),
+            "can_write": repo.can_write(),
+            "name": repo.get_name().await.unwrap_or_default(),
+            "refreshed_files": json!(Vec::<String>::new()), // Initialize empty
+            "all_files": json!(Vec::<String>::new())       // Initialize empty
+        });
+        let mut refreshed_files_vec = Vec::new();
+        let mut all_files_vec = Vec::new();
+
+        // Get current repo hash and collection info
+        match repo.get_hash_from_dht().await {
+            Ok(repo_hash) => {
+                repo_info["repo_hash"] = json!(repo_hash.to_string());
+                
+                // Refresh collection hash if needed
+                log_debug!(TAG, "Repo {} has DHT hash {}. Checking if group has it locally.", repo.id(), repo_hash);
+                if !group.has_hash(&repo_hash).await? {
+                    log_debug!(TAG, "Repo {} collection {} not found locally. Downloading...", repo.id(), repo_hash);
+                    match group.download_hash_from_peers(&repo_hash).await {
+                        Ok(_) => {
+                            log_debug!(TAG, "Successfully downloaded collection hash {} for repo {}", repo_hash, repo.id());
+                        }
+                        Err(e) => {
+                            log_debug!(TAG, "Error downloading collection hash {} for repo {}: {}", repo_hash, repo.id(), e);
+                            repo_info["error"] = json!(format!("Error downloading collection: {}", e));
+                            refreshed_repos.push(repo_info);
+                            continue; // Skip to next repo if download fails
+                        }
                     }
+                } else {
+                    log_debug!(TAG, "Repo {} collection {} already local.", repo.id(), repo_hash);
                 }
-                Err(e) => {
-                    log_debug!(TAG, "Error getting hash for file {}: {}", file_name, e);
+
+                // Now that the collection is ensured to be local, list all files in the repo
+                match repo.list_files().await {
+                    Ok(files) => {
+                        log_debug!(TAG, "Repo {} lists files: {:?}", repo.id(), files);
+                        all_files_vec = files;
+                    }
+                    Err(e) => {
+                        log_debug!(TAG, "Error listing files for repo {} after ensuring collection download: {}", repo.id(), e);
+                        // Even if listing fails here, we might have a repo_hash, so continue with empty files.
+                        // Or, handle as a more significant error. For now, log and continue.
+                        repo_info["error_listing_files"] = json!(format!("Error listing files post-download: {}", e));
+                    }
+                };
+                repo_info["all_files"] = json!(all_files_vec.clone());
+
+
+                // For each file, check if it needs to be refreshed
+                for file_name in &all_files_vec {
+                    match repo.get_file_hash(file_name).await {
+                        Ok(file_hash) => {
+                            if !group.has_hash(&file_hash).await? {
+                                log_debug!(TAG, "File {} hash {} not found locally. Downloading...", file_name, file_hash);
+                                match group.download_hash_from_peers(&file_hash).await {
+                                    Ok(_) => {
+                                        log_debug!(TAG, "Successfully downloaded file hash {} for {}", file_hash, file_name);
+                                        refreshed_files_vec.push(file_name.clone());
+                                    }
+                                    Err(e) => {
+                                        log_debug!(TAG, "Error downloading file {} hash {}: {}", file_name, file_hash, e);
+                                        // Optionally add to a list of files that failed to download
+                                    }
+                                }
+                            }
+                        }
+                        Err(e) => {
+                            log_debug!(TAG, "Error getting hash for file {}: {}", file_name, e);
+                        }
+                    }
                 }
+                repo_info["refreshed_files"] = json!(refreshed_files_vec);
+            }
+            Err(e) => {
+                log_debug!(TAG, "Error getting repo hash for {}: {}", repo.id(), e);
+                repo_info["error"] = json!(format!("Error getting repo hash from DHT: {}", e));
             }
         }
+
+        refreshed_repos.push(repo_info);
     }
 
     Ok(HttpResponse::Ok().json(json!({
         "status": "success",
-        "refreshed_files": refreshed_files
+        "repos": refreshed_repos
     })))
 }