pispas_modules/

mypos.rs

use std::process::{Child, Command};
use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use std::sync::{
    atomic::{AtomicBool, Ordering},
    Arc,
};
use tokio::sync::{Mutex, oneshot};
use tokio::time::{Duration, timeout, sleep};
use crate::service::{Service, WebSocketWrite};
use easy_trace::prelude::{info, error, warn};
use tokio_tungstenite::{connect_async, tungstenite::protocol::Message, MaybeTlsStream, WebSocketStream};
use futures_util::{stream::SplitSink, SinkExt, StreamExt, stream::SplitStream};
use tokio::net::TcpStream;

// Imports específicos de Windows
#[cfg(target_os = "windows")]
use {sharing::MYPOS_SERVER_FILE, sharing::proc::kill_process_by_name};

/// MyPOS service version
pub const MYPOS_VERSION: &str = "1.1.0";

/// Número máximo de reintentos para operaciones
const MAX_RETRIES: u32 = 3;

/// Tiempo de espera entre reintentos (en segundos)
const RETRY_DELAY_SECS: u64 = 2;

/// Intervalo de verificación del watchdog (en segundos)
const WATCHDOG_CHECK_INTERVAL_SECS: u64 = 5;

/// Tiempo de espera antes de reiniciar el servidor después de un fallo (en segundos)
const RESTART_DELAY_SECS: u64 = 3;

/// Tiempo máximo de espera para la conexión WebSocket (en segundos)
const WEBSOCKET_CONNECT_TIMEOUT_SECS: u64 = 10;

#[derive(Debug, Serialize, Deserialize)]
struct ServerResponse {
    status: String,
    message: Option<String>,
}

type WsWriter = SplitSink<WebSocketStream<MaybeTlsStream<TcpStream>>, Message>;
type WsReader = SplitStream<WebSocketStream<MaybeTlsStream<TcpStream>>>;

/// MyPOS Service implementation with automatic retries and watchdog monitoring
pub struct MyPOSService {
    /// Handle for the `mypos_server.exe` process (solo Windows).
    #[cfg(target_os = "windows")]
    child_process: Arc<Mutex<Option<Child>>>,

    /// State indicating if the service is running.
    running: Arc<AtomicBool>,

    /// State indicating if watchdog should be active
    watchdog_enabled: Arc<AtomicBool>,

    /// WebSocket writer for communicating with the Python server
    ws_writer: Arc<Mutex<Option<WsWriter>>>,

    /// Channel for sending responses (store the sender, not receiver)
    response_sender: Arc<Mutex<Option<oneshot::Sender<ServerResponse>>>>,

    /// Counter for restart attempts
    restart_count: Arc<Mutex<u32>>,
}

impl MyPOSService {
    /// Creates a new instance of the MyPOSService.
    pub fn new() -> Self {
        MyPOSService {
            #[cfg(target_os = "windows")]
            child_process: Arc::new(Mutex::new(None)),
            running: Arc::new(AtomicBool::new(false)),
            watchdog_enabled: Arc::new(AtomicBool::new(false)),
            ws_writer: Arc::new(Mutex::new(None)),
            response_sender: Arc::new(Mutex::new(None)),
            restart_count: Arc::new(Mutex::new(0)),
        }
    }

    // ========================================================================
    // SECCIÓN: GESTIÓN DEL PROCESO Y CONEXIÓN
    // ========================================================================

    /// Launches the MyPOS server process (Windows only)
    #[cfg(target_os = "windows")]
    async fn launch_process(&self) -> Result<(), String> {
        let exe_path = sharing::get_path_mypos_server();
        info!("Launching MyPOS server at: {}", exe_path);

        let mut guard = self.child_process.lock().await;

        match Command::new(exe_path).spawn() {
            Ok(child) => {
                *guard = Some(child);
                info!("MyPOS server process started successfully.");
                Ok(())
            }
            Err(e) => {
                error!("Failed to start MyPOS server: {:?}", e);
                Err(format!("Failed to start process: {}", e))
            }
        }
    }

    /// Connects to the WebSocket server with timeout
    async fn connect_websocket(&self) -> Result<(), String> {
        let url = "ws://127.0.0.1:8765";
        info!("Connecting to WebSocket at: {}", url);

        let connect_future = connect_async(url);

        let (ws_stream, _) = match timeout(
            Duration::from_secs(WEBSOCKET_CONNECT_TIMEOUT_SECS),
            connect_future
        ).await {
            Ok(Ok(result)) => result,
            Ok(Err(e)) => return Err(format!("Failed to connect to WebSocket: {}", e)),
            Err(_) => return Err("WebSocket connection timeout".to_string()),
        };

        let (write, read) = ws_stream.split();

        // Store the writer
        *self.ws_writer.lock().await = Some(write);

        // Start reading responses
        let response_sender = self.response_sender.clone();
        tokio::spawn(async move {
            Self::handle_websocket_responses(read, response_sender).await;
        });

        info!("Connected to WebSocket successfully");
        Ok(())
    }

    /// Handles incoming WebSocket responses
    async fn handle_websocket_responses(
        mut read: WsReader,
        response_sender: Arc<Mutex<Option<oneshot::Sender<ServerResponse>>>>
    ) {
        while let Some(msg_result) = read.next().await {
            match msg_result {
                Ok(Message::Text(response)) => {
                    if let Ok(server_response) = serde_json::from_str::<ServerResponse>(&response) {
                        info!("Received response: {:?}", server_response);

                        // Send to the current waiting sender
                        let mut sender_guard = response_sender.lock().await;
                        if let Some(sender) = sender_guard.take() {
                            if sender.send(server_response).is_err() {
                                warn!("Failed to send response to waiting transaction");
                            }
                        }
                    } else {
                        warn!("Received unexpected response: {}", response);
                    }
                }
                Ok(Message::Close(_)) => {
                    warn!("WebSocket connection closed");
                    break;
                }
                Err(e) => {
                    error!("WebSocket error: {:?}", e);
                    break;
                }
                _ => {}
            }
        }
        info!("WebSocket response handler terminated");
    }

    // ========================================================================
    // SECCIÓN: WATCHDOG Y MONITOREO
    // ========================================================================

    /// Launches and monitors the MyPOS server with watchdog (Windows)
    #[cfg(target_os = "windows")]
    async fn launch_and_monitor(&self) {
        // Habilitar watchdog
        self.watchdog_enabled.store(true, Ordering::Relaxed);

        // Intentar lanzar el proceso
        if let Err(e) = self.launch_process().await {
            error!("Failed to launch process: {}", e);
            return;
        }

        self.running.store(true, Ordering::Relaxed);

        // Wait for the server to start
        sleep(Duration::from_secs(RESTART_DELAY_SECS)).await;

        // Connect to WebSocket
        if let Err(e) = self.connect_websocket().await {
            error!("Failed to connect to WebSocket: {:?}", e);
            return;
        }

        // Iniciar watchdog
        self.start_watchdog().await;
    }

    /// Launches connection monitoring for macOS (no process management)
    #[cfg(target_os = "macos")]
    async fn launch_and_monitor(&self) {
        info!("macOS: Skipping process launch, connecting to existing WebSocket server...");

        self.running.store(true, Ordering::Relaxed);
        self.watchdog_enabled.store(true, Ordering::Relaxed);

        // Intentar conectar al WebSocket
        if let Err(e) = self.connect_websocket().await {
            error!("Failed to connect to WebSocket: {:?}", e);
            self.running.store(false, Ordering::Relaxed);
            return;
        }

        info!("macOS: Connected to existing WebSocket server");

        // Iniciar watchdog solo para monitorear la conexión
        self.start_watchdog().await;
    }

    /// Starts the watchdog that monitors the process and connection health
    async fn start_watchdog(&self) {
        let running = self.running.clone();
        let watchdog_enabled = self.watchdog_enabled.clone();
        let ws_writer = self.ws_writer.clone();
        let restart_count = self.restart_count.clone();
        let response_sender = self.response_sender.clone();

        #[cfg(target_os = "windows")]
        let child_process = self.child_process.clone();

        tokio::spawn(async move {
            info!("Watchdog started");

            while watchdog_enabled.load(Ordering::Relaxed) {
                sleep(Duration::from_secs(WATCHDOG_CHECK_INTERVAL_SECS)).await;

                if !watchdog_enabled.load(Ordering::Relaxed) {
                    info!("Watchdog disabled, stopping...");
                    break;
                }

                let mut needs_restart = false;
                let mut process_died = false;

                // Verificar si el proceso sigue corriendo (solo Windows)
                #[cfg(target_os = "windows")]
                {
                    let mut child_guard = child_process.lock().await;

                    if let Some(child) = child_guard.as_mut() {
                        match child.try_wait() {
                            Ok(Some(status)) => {
                                // Proceso murió
                                error!("MyPOS server process exited with status: {:?}", status);
                                *child_guard = None;
                                drop(child_guard);
                                needs_restart = true;
                                process_died = true;
                            }
                            Ok(None) => {
                                // Proceso sigue corriendo
                                drop(child_guard);
                            }
                            Err(e) => {
                                // Error al verificar estado
                                error!("Error checking process status: {:?}", e);
                                *child_guard = None;
                                drop(child_guard);
                                needs_restart = true;
                                process_died = true;
                            }
                        }
                    } else {
                        // No hay proceso hijo
                        drop(child_guard);
                        if running.load(Ordering::Relaxed) {
                            warn!("Process handle lost, attempting restart");
                            needs_restart = true;
                            process_died = true;
                        }
                    }
                }

                // Verificar si la conexión WebSocket está activa
                if ws_writer.lock().await.is_none() && running.load(Ordering::Relaxed) {
                    warn!("WebSocket connection lost");
                    needs_restart = true;
                }

                // Si necesita reiniciar
                if needs_restart {
                    let mut count = restart_count.lock().await;
                    *count += 1;
                    let attempt = *count;
                    drop(count);

                    warn!("🔄 Watchdog detected failure. Restart attempt #{}", attempt);

                    // Limpiar estado
                    *ws_writer.lock().await = None;
                    running.store(false, Ordering::Relaxed);

                    // Reintentar solo en Windows
                    #[cfg(target_os = "windows")]
                    {
                        info!("Watchdog: Attempting to restart MyPOS server...");

                        // Si el proceso murió, limpiar procesos residuales
                        if process_died {
                            info!("Watchdog: Cleaning up residual processes...");
                            kill_process_by_name(MYPOS_SERVER_FILE);
                            sleep(Duration::from_millis(500)).await;
                        }

                        // Esperar antes de reintentar
                        sleep(Duration::from_secs(RESTART_DELAY_SECS)).await;

                        // Lanzar proceso nuevamente
                        let exe_path = sharing::get_path_mypos_server();
                        info!("Watchdog: Spawning new process at: {}", exe_path);

                        match Command::new(&exe_path).spawn() {
                            Ok(child) => {
                                *child_process.lock().await = Some(child);
                                info!("Watchdog: ✓ Process restarted successfully");

                                // Esperar a que el servidor inicie
                                info!("Watchdog: Waiting {} seconds for server to start...", RESTART_DELAY_SECS);
                                sleep(Duration::from_secs(RESTART_DELAY_SECS)).await;

                                // Reconectar WebSocket
                                info!("Watchdog: Attempting to reconnect WebSocket...");
                                match Self::reconnect_websocket_static(ws_writer.clone(), response_sender.clone()).await {
                                    Ok(_) => {
                                        running.store(true, Ordering::Relaxed);
                                        info!("Watchdog: ✓ WebSocket reconnected successfully");
                                        info!("Watchdog: ✓ Full recovery completed on attempt #{}", attempt);
                                    }
                                    Err(e) => {
                                        error!("Watchdog: ✗ Failed to reconnect WebSocket: {}", e);
                                    }
                                }
                            }
                            Err(e) => {
                                error!("Watchdog: ✗ Failed to restart process: {:?}", e);
                            }
                        }
                    }

                    // En macOS solo intentar reconectar
                    #[cfg(target_os = "macos")]
                    {
                        sleep(Duration::from_secs(RESTART_DELAY_SECS)).await;
                        info!("Watchdog: Attempting to reconnect WebSocket...");
                        match Self::reconnect_websocket_static(ws_writer.clone(), response_sender.clone()).await {
                            Ok(_) => {
                                running.store(true, Ordering::Relaxed);
                                info!("Watchdog: WebSocket reconnected successfully");
                            }
                            Err(e) => {
                                error!("Watchdog: Failed to reconnect: {}", e);
                            }
                        }
                    }
                }
            }

            info!("Watchdog stopped");
        });
    }

    /// Helper estático para reconectar WebSocket desde el watchdog
    async fn reconnect_websocket_static(
        ws_writer: Arc<Mutex<Option<WsWriter>>>,
        response_sender: Arc<Mutex<Option<oneshot::Sender<ServerResponse>>>>
    ) -> Result<(), String> {
        let url = "ws://127.0.0.1:8765";
        info!("Reconnecting to WebSocket at: {}", url);

        let connect_future = connect_async(url);
        let (ws_stream, _) = match timeout(
            Duration::from_secs(WEBSOCKET_CONNECT_TIMEOUT_SECS),
            connect_future
        ).await {
            Ok(Ok(result)) => result,
            Ok(Err(e)) => return Err(format!("Failed to reconnect: {}", e)),
            Err(_) => return Err("Reconnection timeout".to_string()),
        };

        let (write, read) = ws_stream.split();
        *ws_writer.lock().await = Some(write);

        // Restart the WebSocket response handler
        tokio::spawn(async move {
            Self::handle_websocket_responses(read, response_sender).await;
        });

        info!("WebSocket reconnection completed");
        Ok(())
    }

    // ========================================================================
    // SECCIÓN: DETENCIÓN DEL SERVICIO
    // ========================================================================

    /// Stops the MyPOS server (Windows)
    #[cfg(target_os = "windows")]
    async fn stop(&self) {
        info!("Stopping MyPOS service...");

        // Deshabilitar watchdog primero
        self.watchdog_enabled.store(false, Ordering::Relaxed);

        // Esperar un momento para que el watchdog termine
        sleep(Duration::from_millis(500)).await;

        // Close WebSocket connection
        *self.ws_writer.lock().await = None;

        // Stop process
        let mut guard = self.child_process.lock().await;
        if let Some(mut child) = guard.take() {
            if let Err(e) = child.kill() {
                error!("Error killing MyPOS server: {:?}", e);
            } else {
                kill_process_by_name(MYPOS_SERVER_FILE);
                info!("MyPOS server stopped.");
            }
        }

        self.running.store(false, Ordering::Relaxed);
        *self.restart_count.lock().await = 0;
    }

    /// Stops the service (macOS - only closes WebSocket)
    #[cfg(target_os = "macos")]
    async fn stop(&self) {
        info!("macOS: Closing WebSocket connection...");

        // Deshabilitar watchdog
        self.watchdog_enabled.store(false, Ordering::Relaxed);

        // Esperar un momento
        sleep(Duration::from_millis(500)).await;

        // Close WebSocket connection
        *self.ws_writer.lock().await = None;

        self.running.store(false, Ordering::Relaxed);
        *self.restart_count.lock().await = 0;
        info!("MyPOS WebSocket connection closed.");
    }

    // ========================================================================
    // SECCIÓN: COMUNICACIÓN CON EL SERVIDOR
    // ========================================================================

    /// Sends a message to the WebSocket server and waits for response
    async fn send_websocket_message(&self, message: Value) -> Result<ServerResponse, String> {
        // Create a channel for the response
        let (sender, receiver) = oneshot::channel();

        // Store the sender for this transaction
        {
            let mut response_sender = self.response_sender.lock().await;
            *response_sender = Some(sender);
        }

        // Send the message
        {
            let mut ws_writer = self.ws_writer.lock().await;
            if let Some(writer) = ws_writer.as_mut() {
                let msg = Message::Text(message.to_string());
                if let Err(e) = writer.send(msg).await {
                    return Err(format!("Failed to send WebSocket message: {}", e));
                }
            } else {
                return Err("WebSocket not connected".to_string());
            }
        }

        // Wait for response with timeout
        match timeout(Duration::from_secs(60), receiver).await {
            Ok(Ok(response)) => Ok(response),
            Ok(Err(_)) => Err("Response channel closed".to_string()),
            Err(_) => Err("Transaction timeout".to_string()),
        }
    }

    // ========================================================================
    // SECCIÓN: REINTENTOS AUTOMÁTICOS
    // ========================================================================

    /// Attempts a transaction with automatic retries
    async fn attempt_transaction_with_retries(
        &self,
        action: &str,
        amount: f64,
    ) -> Result<ServerResponse, String> {
        for attempt in 1..=MAX_RETRIES {
            info!("Attempt {}/{} for {} of {:.2} EUR", attempt, MAX_RETRIES, action, amount);

            let message = json!({
                "action": action,
                "amount": amount
            });

            match self.send_websocket_message(message).await {
                Ok(response) => {
                    let status_lower = response.status.to_lowercase();

                    // Si es exitoso, retornar inmediatamente
                    if status_lower == "success" {
                        info!("Transaction successful on attempt {}", attempt);
                        return Ok(response);
                    }

                    // Si fue cancelado por el usuario, no reintentar
                    if status_lower == "usercancel" {
                        warn!("Transaction cancelled by user, no retries");
                        return Ok(response);
                    }

                    // Para otros errores, continuar con reintentos si quedan intentos
                    if attempt < MAX_RETRIES {
                        warn!("Attempt {} failed with status: {}. Retrying in {} seconds...",
                              attempt, response.status, RETRY_DELAY_SECS);
                        sleep(Duration::from_secs(RETRY_DELAY_SECS)).await;
                    } else {
                        error!("All {} attempts failed. Last status: {}", MAX_RETRIES, response.status);
                        return Ok(response);
                    }
                }
                Err(e) => {
                    if attempt < MAX_RETRIES {
                        warn!("Attempt {} failed with error: {}. Retrying in {} seconds...",
                              attempt, e, RETRY_DELAY_SECS);
                        sleep(Duration::from_secs(RETRY_DELAY_SECS)).await;
                    } else {
                        error!("All {} attempts failed. Last error: {}", MAX_RETRIES, e);
                        return Err(e);
                    }
                }
            }
        }

        Err("Maximum retries exceeded".to_string())
    }

    // ========================================================================
    // SECCIÓN: MANEJO DE COMPRAS Y DEVOLUCIONES
    // ========================================================================

    /// Handles a purchase with automatic retries
    async fn handle_purchase(&self, amount: f64, write: WebSocketWrite) -> (i32, String) {
        info!("Handling purchase of: {:.2} EUR (with up to {} retries)", amount, MAX_RETRIES);

        match self.attempt_transaction_with_retries("purchase", amount).await {
            Ok(response) => {
                let success = response.status.to_lowercase() == "success";
                let message = response.message.unwrap_or_else(||
                    if success {
                        format!("Purchase of {:.2} EUR completed successfully", amount)
                    } else {
                        format!("Purchase of {:.2} EUR failed after {} attempts: {}",
                                amount, MAX_RETRIES, response.status)
                    }
                );

                // Send response to the main WebSocket client if available
                if let Some(ws_lock) = &write {
                    let response_msg = json!({
                        "type": "purchase_result",
                        "success": success,
                        "amount": amount,
                        "message": message,
                        "status": response.status
                    });

                    let mut ws = ws_lock.write().await;
                    if let Err(e) = ws.send(Message::Text(response_msg.to_string())).await {
                        error!("Failed to send purchase result: {:?}", e);
                    }
                }

                if success { (0, message) } else { (1, message) }
            }
            Err(e) => {
                let error_msg = format!("Purchase failed after {} attempts: {}", MAX_RETRIES, e);
                error!("{}", error_msg);
                (1, error_msg)
            }
        }
    }

    /// Handles a refund with automatic retries
    async fn handle_refund(&self, amount: f64, write: WebSocketWrite) -> (i32, String) {
        info!("Handling refund of: {:.2} EUR (with up to {} retries)", amount, MAX_RETRIES);

        match self.attempt_transaction_with_retries("refund", amount).await {
            Ok(response) => {
                let success = response.status.to_lowercase() == "success";
                let message = response.message.unwrap_or_else(||
                    if success {
                        format!("Refund of {:.2} EUR completed successfully", amount)
                    } else {
                        format!("Refund of {:.2} EUR failed after {} attempts: {}",
                                amount, MAX_RETRIES, response.status)
                    }
                );

                // Send response to the main WebSocket client if available
                if let Some(ws_lock) = &write {
                    let response_msg = json!({
                        "type": "refund_result",
                        "success": success,
                        "amount": amount,
                        "message": message,
                        "status": response.status
                    });

                    let mut ws = ws_lock.write().await;
                    if let Err(e) = ws.send(Message::Text(response_msg.to_string())).await {
                        error!("Failed to send refund result: {:?}", e);
                    }
                }

                if success { (0, message) } else { (1, message) }
            }
            Err(e) => {
                let error_msg = format!("Refund failed after {} attempts: {}", MAX_RETRIES, e);
                error!("{}", error_msg);
                (1, error_msg)
            }
        }
    }
}

// ========================================================================
// IMPLEMENTACIÓN DEL TRAIT SERVICE
// ========================================================================

#[async_trait]
impl Service for MyPOSService {
    /// Handles incoming service actions (START, STOP, PURCHASE, REFUND)
    async fn run(&self, action: Value, write: WebSocketWrite) -> (i32, String) {
        info!("MyPOSService: Running action: {:?}", action);

        // Parse ACTION similar to ScaleService
        let parsed_action = if let Some(action_str) = action.get("ACTION").and_then(|a| a.as_str()) {
            // If ACTION is a string, parse it as JSON
            serde_json::from_str::<Value>(action_str).unwrap_or_else(|_| {
                // If parsing fails, treat the string as a simple command
                json!({"command": action_str})
            })
        } else if let Some(action_obj) = action.get("ACTION") {
            // If ACTION is already an object, use it directly
            action_obj.clone()
        } else {
            // Fallback: use the action itself
            action
        };

        let command = parsed_action
            .get("command")
            .and_then(|c| c.as_str())
            .unwrap_or("UNKNOWN");

        info!("MyPOS Command: {}", command);

        match command {
            "START" => {
                self.launch_and_monitor().await;

                #[cfg(target_os = "windows")]
                let msg = "MyPOS server started with watchdog monitoring enabled.";

                #[cfg(target_os = "macos")]
                let msg = "MyPOS WebSocket connection established with watchdog monitoring (no process launched on macOS).";

                (0, msg.to_string())
            }
            "STOP" => {
                self.stop().await;
                (0, "MyPOS stopped.".to_string())
            }
            "PURCHASE" => {
                let amount = parsed_action.get("amount").and_then(|a| a.as_f64()).unwrap_or(0.0);
                if amount > 0.0 {
                    self.handle_purchase(amount, write).await
                } else {
                    (1, "Invalid purchase amount.".to_string())
                }
            }
            "REFUND" => {
                let amount = parsed_action.get("amount").and_then(|a| a.as_f64()).unwrap_or(0.0);
                if amount > 0.0 {
                    self.handle_refund(amount, write).await
                } else {
                    (1, "Invalid refund amount.".to_string())
                }
            }
            _ => (1, format!("Unknown command: {}", command)),
        }
    }

    /// Converts the service instance into a `dyn Any` reference
    fn as_any(&self) -> &dyn std::any::Any {
        self
    }

    /// Stops the service and cleans up any running processes
    fn stop_service(&self) {
        let running = self.running.clone();
        let watchdog_enabled = self.watchdog_enabled.clone();

        #[cfg(target_os = "windows")]
        let child_process = self.child_process.clone();

        let ws_writer = self.ws_writer.clone();

        tokio::spawn(async move {
            if running.load(Ordering::Relaxed) {
                // Deshabilitar watchdog
                watchdog_enabled.store(false, Ordering::Relaxed);
                sleep(Duration::from_millis(500)).await;

                // Close WebSocket
                *ws_writer.lock().await = None;

                // Stop process (solo Windows)
                #[cfg(target_os = "windows")]
                {
                    if let Some(mut child) = child_process.lock().await.take() {
                        if let Err(e) = child.kill() {
                            error!("Error stopping MyPOS server: {:?}", e);
                        } else {
                            info!("MyPOS server stopped cleanly.");
                        }
                    }
                }

                #[cfg(target_os = "macos")]
                info!("MyPOS WebSocket closed (no process to stop on macOS).");

                running.store(false, Ordering::Relaxed);
            }
        });
    }

    /// Returns the version of the MyPOS service
    fn get_version(&self) -> String {
        MYPOS_VERSION.to_string()
    }
}