v1.0 with SW PWA enabled

node/src/main.rs

@@ -0,0 +1,417 @@
+// src/main.rs
+use std::{fs, time::Duration};
+use serde::{Serialize, Deserialize};
+use std::process::Command;
+use std::sync::{Arc};
+use axum::{routing::post, routing::get, routing::delete, Json, Router, extract::Path, http::Method, http::HeaderValue};
+use tokio::task;
+use tokio::time::{self, interval};
+use tokio::sync::Mutex as AsyncMutex;
+use clap::{Parser, Subcommand};
+use reqwest;
+use chrono::{DateTime, Utc};
+use std::env;
+use tower_http::cors::{Any, CorsLayer};
+
+const GO_SERVER_URL: &str = "http://192.168.1.75:8082/api"; 
+const NODE_NAME: &str = "MyRustNode-01";
+const NODE_PORT: &str = "8081";
+
+#[derive(Deserialize)]
+struct DeleteRequest {
+    service_id: i32,
+}
+
+#[derive(Parser, Debug)]
+#[command(author, version, about, long_about = None)]
+struct Cli {
+    #[command(subcommand)]
+    command: Option<Commands>,
+}
+
+#[derive(Subcommand, Debug)]
+enum Commands {
+    AddProcess {
+        #[arg(short, long)]
+        name: String,
+
+        #[arg(short, long)]
+        cmd: String,
+    },
+    StartMonitor,
+}
+
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct StatusRecord {
+    pub timestamp: DateTime<Utc>,
+    pub status: u8,
+}
+
+impl Default for StatusRecord {
+    fn default() -> Self {
+        StatusRecord {
+            timestamp: Utc::now(), // Vec::new() est Vec::default()
+            status: 0,
+        }
+    }
+}
+
+#[derive(Debug, Serialize, Deserialize, Clone)]
+struct ProcessConfig {
+    #[serde(default)]
+    id: i32,
+    name: String,
+    command: String,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct FileData {
+    node_id: i32,
+    processes: Vec<ProcessConfig>,
+}
+
+static PROCESSES_FILE: &str = "processes.json";
+
+async fn list_processes(state: axum::extract::State<AppState>) -> Json<Vec<ProcessConfig>> {
+    let guard = state.processes.lock().await;
+    Json(guard.clone())
+}
+
+#[derive(Clone)]
+struct AppState {
+    node_id: Arc<AsyncMutex<i32>>,
+    processes: Arc<AsyncMutex<Vec<ProcessConfig>>>,
+}
+
+#[derive(Serialize)]
+struct RegistrationRequest {
+    id: i32,
+    name: String,
+    address: String,
+}
+
+#[derive(Serialize)]
+struct RegistrationServiceRequest {
+    node_id: i32,
+    service: ProcessConfig,
+}
+
+#[derive(Deserialize)]
+struct RegisterResponse {
+    id: i32,
+}
+
+#[derive(Serialize)]
+struct ServiceUpdateRequest {
+    service_id: i32,
+    service_status: StatusRecord,
+}
+
+#[derive(Serialize)]
+struct UpdateRequest {
+    node_id: String,
+    #[serde(default)]
+    services: Vec<ServiceUpdateRequest>,
+}
+
+async fn add_process(
+    state: axum::extract::State<AppState>,
+    Json(payload): Json<ProcessConfig>,
+) -> Json<String> {
+    let mut vec_guard: tokio::sync::MutexGuard<'_, Vec<ProcessConfig>> = state.processes.lock().await;
+    let id_guard: tokio::sync::MutexGuard<'_, i32> = state.node_id.lock().await;
+
+    let mut new_process = payload.clone();
+
+    let client = reqwest::Client::new();
+    let request_payload = RegistrationServiceRequest {
+        node_id: id_guard.clone(),
+        service: new_process.clone(),
+    };
+    
+    match client
+        .post(format!("{}/registerService", GO_SERVER_URL))
+        .json(&request_payload)
+        .send()
+        .await
+    {
+        Ok(response) => {
+            if response.status().is_success() {
+                match response.json::<RegisterResponse>().await {
+                    Ok(data) => {
+                        new_process.id = data.id;
+                        vec_guard.push(new_process.clone());
+                        save_to_disk(&FileData { node_id: data.id, processes: vec_guard.clone() });
+
+                        return Json(format!("✅ Enregistrement réussi. ID du service attribué : {}", data.id));
+                    }
+                    Err(_) => { return Json(format!("Process error on json received")); },
+                }
+            } else {
+                return Json(format!("⚠️ Échec de l'enregistrement, statut: {}", response.status())); 
+            }
+        }
+        Err(e) => {
+            return Json(format!("❌ Impossible de joindre le serveur Go : {}. Nouvelle tentative dans 5s...", e)); 
+        }
+    }
+}
+
+async fn delete_service(
+    state: axum::extract::State<AppState>,
+    Json(payload): Json<DeleteRequest>,
+) -> impl axum::response::IntoResponse {
+
+    let mut guard = state.processes.lock().await;
+    let id_guard: tokio::sync::MutexGuard<'_, i32> = state.node_id.lock().await;
+    let service_id = payload.service_id;
+    
+    let index_to_remove = guard.iter().position(|service| {
+        // service est de type &Service
+        service.id == service_id
+    });
+
+    match index_to_remove {
+        Some(index) => {
+            // ÉTAPE 2: La suppression
+            // On n'emprunte 'guard' qu'ici, après que l'itérateur soit terminé.
+            let removed_service = guard.remove(index);
+            
+            println!("Service supprimé avec succès: {}", removed_service.name);
+
+            save_to_disk(&FileData { node_id: id_guard.clone(), processes: guard.clone() });
+
+            return axum::http::StatusCode::NO_CONTENT;
+            // Retourner removed_service ou un statut de succès
+        }
+        None => {
+            // L'élément n'a pas été trouvé
+            println!("Erreur: Service ID {} non trouvé.", service_id);
+
+            return axum::http::StatusCode::NOT_FOUND;
+        }
+    }
+}
+
+async fn register_with_server(state: AppState) {
+
+    let mut id_guard: tokio::sync::MutexGuard<'_, i32> = state.node_id.lock().await;
+    let guard = state.processes.lock().await;
+
+    let host_ip = env::var("HOST_IP").unwrap_or_else(|_| String::from("0.0.0.0"));
+
+    let address_full = format!( "http://{}:{}", host_ip, NODE_PORT );
+
+    eprintln!("full ip is {}", address_full);
+
+    let client = reqwest::Client::new();
+    let payload = RegistrationRequest {
+        id: id_guard.clone(),
+        name: NODE_NAME.to_string(),
+        address: address_full,
+    };
+
+// Tenter la déclaration (réessayer si le serveur n'est pas prêt)
+    loop {
+        match client
+            .post(format!("{}/register", GO_SERVER_URL))
+            .json(&payload)
+            .send()
+            .await
+        {
+            Ok(response) => {
+                if response.status().is_success() {
+                    match response.json::<RegisterResponse>().await {
+                        Ok(data) => {
+                            println!("✅ Enregistrement réussi. ID du node attribué : {}", data.id);
+                            *id_guard = data.id;
+
+                            save_to_disk(&FileData { node_id: data.id, processes: guard.clone() });
+                        }
+                        Err(_) => eprintln!("⚠️ Réponse réussie mais format JSON invalide"),
+                    }
+                    break;
+                } else {
+                    eprintln!("⚠️ Échec de l'enregistrement, statut: {}", response.status());
+                }
+            }
+            Err(e) => {
+                eprintln!("❌ Impossible de joindre le serveur Go : {}. Nouvelle tentative dans 5s...", e);
+            }
+        }
+        time::sleep(Duration::from_secs(5)).await;
+    }
+
+}
+
+fn save_to_disk(data: &FileData) {
+    let json = serde_json::to_string_pretty(&data).unwrap();
+    fs::write(PROCESSES_FILE, json).unwrap();
+}
+
+fn initialize_processes_file() {
+    let path = "processes.json";
+
+    if !std::path::Path::new(path).exists() {
+        println!("No processes.json found, creating an empty one...");
+        std::fs::write(path, "[]").expect("Failed to create processes.json");
+    }
+    else
+    {
+        println!("processes.json found");
+    }
+}
+
+fn load_from_disk() -> FileData{
+    if let Ok(content) = fs::read_to_string(PROCESSES_FILE) {
+        if let Ok(parsed) = serde_json::from_str::<FileData>(&content) {
+            return parsed;
+        }
+    }
+    FileData{ node_id: 0, processes: vec![] }
+}
+
+fn check_process_running(cmd: &str) -> bool {
+    let output_result = Command::new("sh").arg("-c").arg(cmd).output();
+
+    match output_result {
+        Ok(output) => {
+            // S'assurer que le processus a techniquement réussi (exit code 0), 
+            // bien que nous nous concentrions sur le stdout
+            if !output.status.success() {
+                let stdout_str = match str::from_utf8(&output.stderr) {
+                    Ok(s) => s, // Supprimer les espaces et retours à la ligne
+                    Err(_) => "",
+                };
+
+                println!("error: {}",stdout_str.to_string());
+
+                return false;
+            }
+
+            // 2. Convertir stdout (Vec<u8>) en chaîne de caractères
+            let stdout_str = match str::from_utf8(&output.stdout) {
+                Ok(s) => s.trim(), // Supprimer les espaces et retours à la ligne
+                Err(_) => {
+                    return false;
+                }
+            };
+
+            println!("result: {}",stdout_str.to_string());
+
+            // 3. Analyser la chaîne
+            if stdout_str == "1" {
+                return true;
+            } else if stdout_str == "0" {
+                return false;
+            } else {
+                // La sortie n'est pas celle attendue ("0" ou "1")
+                return false;
+            }
+        }
+        Err(_) => {
+            // Erreur système (ex: sh introuvable ou erreur I/O)
+            return false;
+        }
+    }
+}
+
+async fn start_monitor(state: AppState) {
+
+    let mut interval = interval(Duration::from_mins(1) );
+    interval.tick().await;
+
+    loop {
+        {
+            interval.tick().await;
+
+            let mut update_request = UpdateRequest{ node_id: NODE_NAME.to_string(), services: Vec::new() };
+
+            let mut guard = state.processes.lock().await;
+            println!("--- Checking processes ---");
+            for p in guard.iter_mut() {
+                let running = check_process_running(&p.command);
+                println!("{} => {}", p.name, if running { "RUNNING" } else { "STOPPED" });
+                
+                let mut status = StatusRecord{ status: 0, timestamp: Utc::now() };
+                if running == true {
+                    status.status = 1;
+                }
+
+                update_request.services.push(ServiceUpdateRequest{ service_id: p.id.clone(), service_status: status.clone() });
+            }
+
+            if update_request.services.len() > 0 {
+                let client = reqwest::Client::new();
+                let payload = update_request;
+
+                match client
+                    .post(format!("{}/updateServiceStatus", GO_SERVER_URL ))
+                    .json(&payload)
+                    .send()
+                    .await
+                {
+                    Ok(response) => {
+                        if response.status().is_success() {
+                            println!("✅ Update success");
+                        } else {
+                            eprintln!("⚠️ Update failed {}", response.status());
+                        }
+                    }
+                    Err(e) => {
+                        eprintln!("❌ Error : {}", e);
+                    }
+                }
+            }
+        }
+    }
+}
+
+#[tokio::main]
+async fn main() {
+    initialize_processes_file();
+
+    let initial = load_from_disk();
+
+    println!("initial node_id: {}", initial.node_id);
+
+    let mut state = AppState {
+        node_id: Arc::new(AsyncMutex::new(initial.node_id)),
+        processes: Arc::new(AsyncMutex::new(initial.processes)),
+    };
+
+    let state_monitor = state.clone();
+    task::spawn(async move {
+        start_monitor(state_monitor).await;
+    });
+
+    let state_register = state.clone();
+    tokio::task::spawn(async move {
+        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await; 
+        
+        register_with_server(state_register).await;
+    });
+
+    let cors = CorsLayer::new()
+        // Autorise ton frontend (ex: http://localhost:3000) 
+        // ou Any pour autoriser tout le monde en développement
+        .allow_origin(GO_SERVER_URL.parse::<HeaderValue>().unwrap()) 
+        .allow_methods([Method::GET, Method::POST, Method::DELETE, Method::OPTIONS])
+        .allow_headers(Any);
+
+    let app = Router::new()
+    .route("/add", post(add_process))
+    .route("/list", get(list_processes))
+    .route("/services", delete(delete_service))    
+    .with_state(state)
+    .layer(cors);
+
+    let listener = tokio::net::TcpListener::bind("0.0.0.0:8081").await.unwrap();
+
+    println!("Node service running on port 8081");
+
+    axum::serve(listener,app.into_make_service())
+    .await
+    .unwrap();
+
+}