veza/veza-chat-server/src/main.rs

use axum::extract::ws::{Message as WsMessage, WebSocket};
use axum::{
    extract::{Query, State, WebSocketUpgrade},
    http::StatusCode,
    response::Response,
    routing::{get, post},
    Json, Router,
};
use chat_server::{
    config::SecurityConfig,
    database::pool::create_pool_from_env,
    delivered_status::DeliveredStatusManager, // Add DeliveredStatusManager
    error::ChatError,
    event_bus::RabbitMQEventBus, // Add RabbitMQEventBus import
    jwt_manager::JwtManager,
    models::message::Message,      // Add Message model
    read_receipts::ReadReceiptManager, // Add ReadReceiptManager
    repository::MessageRepository, // Add MessageRepository
    security::permission::PermissionService, // Add PermissionService
    services::MessageEditService, // Add MessageEditService
    typing_indicator::TypingIndicatorManager, // Add TypingIndicatorManager
    // simple_message_store::{SimpleMessage, SimpleMessageStore}, // Remove SimpleMessageStore
    websocket::{
        handler::{websocket_handler, WebSocketState},
        IncomingMessage, OutgoingMessage, WebSocketManager,
    },
};
use futures_util::{FutureExt, SinkExt, StreamExt};
use serde::{Deserialize, Serialize};
use sqlx::PgPool;
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use tokio::net::TcpListener;
use tracing::{error, info, warn};
use uuid::Uuid;

/// État global de l'application
#[derive(Clone)]
struct AppState {
    // store: Arc<SimpleMessageStore>, // Remove SimpleMessageStore
    message_repo: Arc<MessageRepository>, // Add MessageRepository
    _ws_manager: Arc<WebSocketManager>,
    database_pool: Option<sqlx::PgPool>,
    event_bus: Option<Arc<RabbitMQEventBus>>, // Add RabbitMQEventBus, wrapped in Arc for Clone trait
    config: chat_server::config::Config,      // Add Config to AppState
}

/// Requête d'envoi de message
#[derive(Deserialize)]
struct SendMessageRequest {
    conversation_id: Uuid, // Add conversation_id
    content: String,
    sender_id: Uuid, // Use Uuid for sender_id
                     // author: String, // Remove author
                     // room: Option<String>, // Remove room
                     // is_direct: Option<bool>, // Remove is_direct
}

/// Paramètres de récupération de messages
#[derive(Deserialize)]
struct GetMessagesQuery {
    conversation_id: Uuid, // Use conversation_id
    limit: Option<i64>,    // Use i64 for limit
                           // room: Option<String>, // Remove room
                           // user1: Option<String>, // Remove user1
                           // user2: Option<String>, // Remove user2
}

/// Réponse API standard
#[derive(Serialize)]
struct ApiResponse<T> {
    success: bool,
    data: T,
    message: Option<String>,
}

impl<T> ApiResponse<T> {
    fn success(data: T) -> Self {
        Self {
            success: true,
            data,
            message: None,
        }
    }

    fn _error(message: String) -> Self
    where
        T: Default,
    {
        Self {
            success: false,
            data: T::default(),
            message: Some(message),
        }
    }
}

use metrics_exporter_prometheus::PrometheusBuilder;

#[tokio::main]
async fn main() -> Result<(), ChatError> {
    // Configuration du logging avec tracing
    let env_filter = tracing_subscriber::EnvFilter::try_from_default_env()
        .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info"));

    let is_prod = std::env::var("APP_ENV").unwrap_or_default() == "production";

    if is_prod {
        tracing_subscriber::fmt()
            .with_env_filter(env_filter)
            .json()
            .init();
    } else {
        tracing_subscriber::fmt()
            .with_env_filter(env_filter)
            .with_target(true)
            .with_file(true)
            .with_line_number(true)
            .init();
    }

    // Initialisation des métriques Prometheus
    let builder = PrometheusBuilder::new();
    let prometheus_handle = builder
        .install_recorder()
        .map_err(|e| ChatError::configuration_error(&format!("Failed to install Prometheus recorder: {}", e)))?;

    info!("🚀 Démarrage du serveur de chat Veza...");

    let app_config =
        chat_server::config::Config::from_env().map_err(|e| ChatError::Configuration {
            message: e.to_string(),
        })?;

    // Initialisation du pool de connexions à la base de données
    let database_pool = match create_pool_from_env(Some(&app_config.database_url)).await {
        Ok(pool) => {
            info!("✅ Pool de connexions PostgreSQL initialisé avec succès");
            Some(pool)
        }
        Err(e) => {
            warn!("⚠️ Échec d'initialisation du pool de connexions: {}. Le serveur continuera sans base de données.", e);
            None
        }
    };

    // Database pool est requis pour les managers
    let pool_ref = database_pool.as_ref().ok_or_else(|| {
        ChatError::configuration_error("Database pool is required but not initialized")
    })?;
    let message_repo = Arc::new(MessageRepository::new(pool_ref.clone()));
    let read_receipt_manager = Arc::new(ReadReceiptManager::new(pool_ref.clone()));
    let delivered_status_manager = Arc::new(DeliveredStatusManager::new(pool_ref.clone()));
    let typing_indicator_manager = Arc::new(TypingIndicatorManager::new());
    let permission_service = Arc::new(PermissionService::new(pool_ref.clone()));
    let message_edit_service = Arc::new(MessageEditService::new(pool_ref.clone()));

    // Initialisation de l'Event Bus RabbitMQ
    let event_bus = match RabbitMQEventBus::new_with_retry(app_config.rabbit_mq.clone()).await {
        Ok(eb) => {
            info!("✅ Event Bus RabbitMQ initialisé avec succès");
            Some(eb)
        }
        Err(e) => {
            warn!("⚠️ Échec d'initialisation de l'Event Bus RabbitMQ: {}. Le serveur démarrera en mode dégradé (sans Event Bus).", e);
            None
        }
    };

    // Initialisation du gestionnaire WebSocket
    let ws_manager = Arc::new(WebSocketManager::new());

    // Initialisation du gestionnaire JWT
    // SECURITY: JWT_SECRET est REQUIS - pas de valeur par défaut pour éviter les failles de sécurité
    let jwt_secret = chat_server::env::require_env_min_length("JWT_SECRET", 32);

    // SECURITY: Créer SecurityConfig manuellement avec le secret requis
    let security_config = SecurityConfig {
        jwt_secret,
        jwt_access_duration: Duration::from_secs(900), // 15 min
        jwt_refresh_duration: Duration::from_secs(86400 * 30), // 30 days
        jwt_algorithm: "HS256".to_string(),
        jwt_audience: "veza-chat".to_string(),
        jwt_issuer: "veza-backend".to_string(),
        enable_2fa: false,
        totp_window: 1,
        content_filtering: false,
        password_min_length: 8,
        bcrypt_cost: 12,
    };

    // Créer JwtManager avec pool DB si disponible
    let jwt_manager = Arc::new(
        if let Some(ref pool) = database_pool {
            JwtManager::with_pool(security_config, pool.clone())
                .map_err(|e| ChatError::configuration_error(&format!("JWT Manager error: {}", e)))?
        } else {
            JwtManager::new(security_config)
                .map_err(|e| ChatError::configuration_error(&format!("JWT Manager error: {}", e)))?
        }
    );

    // Définir l'adresse d'écoute
    let bind_addr = format!("{}:{}", app_config.host, app_config.port);

    // État pour les routes HTTP (AppState reste pour compatibilité)
    let state = AppState {
        // store: store.clone(), // Remove SimpleMessageStore
        message_repo: message_repo.clone(), // Add MessageRepository
        _ws_manager: ws_manager.clone(),
        database_pool: database_pool.clone(),
        event_bus: event_bus.map(Arc::new), // Add RabbitMQEventBus
        config: app_config.clone(),         // Add app_config to AppState
    };

    // État pour le handler WebSocket
    let ws_state = WebSocketState {
        // store, // Remove SimpleMessageStore
        message_repo: message_repo.clone(), // Add MessageRepository
        read_receipt_manager: read_receipt_manager.clone(), // Add ReadReceiptManager
        delivered_status_manager: delivered_status_manager.clone(), // Add DeliveredStatusManager
        typing_indicator_manager: typing_indicator_manager.clone(), // Add TypingIndicatorManager
        message_edit_service: message_edit_service.clone(), // Add MessageEditService
        ws_manager: ws_manager.clone(),
        jwt_manager,
        permission_service: permission_service.clone(), // Add PermissionService
    };

    // Démarrer le task de monitoring des typing indicators
    // Note: Tokio capture automatiquement les panics dans les tasks spawnées.
    // Toutes les erreurs sont gérées explicitement pour éviter les panics.
    let typing_manager_monitor = typing_indicator_manager.clone();
    let ws_manager_monitor = ws_manager.clone();
    tokio::spawn(async move {
        let mut interval = tokio::time::interval(tokio::time::Duration::from_millis(500));
        loop {
            interval.tick().await;
            
            // Détecter les utilisateurs dont le timeout a expiré
            // Toutes les erreurs sont gérées explicitement pour éviter les panics
            let expired_changes = typing_manager_monitor.monitor_timeouts().await;
            
            // Broadcast les changements de statut (is_typing = false)
            for change in expired_changes {
                let typing_message = OutgoingMessage::UserTyping {
                    conversation_id: change.conversation_id,
                    user_id: change.user_id,
                    is_typing: false,
                };
                
                // Ignorer les erreurs de broadcast pour éviter de bloquer le monitoring
                if let Err(e) = ws_manager_monitor
                    .broadcast_to_conversation(change.conversation_id, typing_message)
                    .await
                {
                    warn!(
                        conversation_id = %change.conversation_id,
                        user_id = %change.user_id,
                        error = %e,
                        "Erreur lors du broadcast de typing timeout"
                    );
                }
            }
        }
    });
    
    info!("✅ Task de monitoring des typing indicators démarré");

    // Configuration des routes avec WebSocket
    let app = Router::new()
        .route("/health", get(health_check))
        .route("/healthz", get(health_check)) // Liveness
        .route("/readyz", get(readiness_check)) // Readiness
        .route(
            "/metrics",
            get(move || std::future::ready(prometheus_handle.render())),
        ) // Prometheus metrics
        .route("/api/messages/{conversation_id}", get(get_messages)) // Update route
        .route("/api/messages", post(send_message))
        .route("/api/messages/stats", get(get_stats))
        .route(
            "/ws",
            get({
                let ws_state_clone = ws_state.clone();
                move |ws: WebSocketUpgrade, query: Query<HashMap<String, String>>| async move {
                    websocket_handler(ws, query, State(ws_state_clone)).await
                }
            }),
        ) // ✨ Handler WebSocket depuis websocket/handler.rs
        .with_state(state); // Utiliser state pour les routes HTTP

    // Démarrage du serveur
    let listener = TcpListener::bind(&bind_addr)
        .await
        .map_err(|e| ChatError::configuration_error(&format!("Bind error on {bind_addr}: {e}")))?;

    info!("✅ Serveur démarré sur http://{}", bind_addr);
    info!("📊 Endpoints disponibles:");
    info!("   - GET  /health          - Vérification de santé");
    info!("   - GET  /api/messages/:conversation_id - Récupération des messages"); // Update endpoint
    info!("   - POST /api/messages    - Envoi de message");
    info!("   - GET  /api/messages/stats - Statistiques");
    info!("   - GET  /ws              - WebSocket Chat (🆕)");

    axum::serve(listener, app)
        .await
        .map_err(|e| ChatError::configuration_error(&format!("Server error: {e}")))?;

    Ok(())
}

/// Endpoint de readiness (DB check)
async fn readiness_check(
    State(state): State<AppState>,
) -> Result<Json<ApiResponse<HashMap<String, String>>>, StatusCode> {
    let mut info = HashMap::new();

    // Check Database
    if let Some(pool) = &state.database_pool {
        if let Err(e) = sqlx::query("SELECT 1").execute(pool).await {
            warn!("Readiness check failed (DB): {}", e);
            return Err(StatusCode::SERVICE_UNAVAILABLE);
        }
    } else {
        warn!("Readiness check failed (No DB pool)");
        return Err(StatusCode::SERVICE_UNAVAILABLE);
    }

    // Check RabbitMQ Event Bus
    if state.config.rabbit_mq.enable {
        if let Some(ref event_bus) = state.event_bus {
            if !event_bus.is_enabled {
                warn!(
                    "Readiness check failed (RabbitMQ EventBus not enabled)"
                );
                return Err(StatusCode::SERVICE_UNAVAILABLE);
            }
        } else {
            warn!(
                "Readiness check failed (RabbitMQ EventBus not initialized but enabled in config)"
            );
            return Err(StatusCode::SERVICE_UNAVAILABLE);
        }
    }

    info.insert("status".to_string(), "ready".to_string());
    Ok(Json(ApiResponse::success(info)))
}

/// Endpoint de vérification de santé
#[tracing::instrument(skip(state))]
async fn health_check(State(state): State<AppState>) -> Json<ApiResponse<HashMap<String, String>>> {
    let mut info = HashMap::new();
    info.insert("status".to_string(), "healthy".to_string());
    info.insert("service".to_string(), "veza-chat-server".to_string());
    info.insert("version".to_string(), "0.3.0".to_string());
    info.insert("websocket".to_string(), "enabled".to_string());

    // Vérification optionnelle de la connexion à la base de données
    if let Some(pool) = &state.database_pool {
        match sqlx::query("SELECT 1").execute(pool).await {
            Ok(_) => {
                info.insert("database".to_string(), "connected".to_string());
            }
            Err(e) => {
                info.insert("database".to_string(), format!("error: {}", e));
                warn!("⚠️ Échec de vérification de la base de données: {}", e);
            }
        }
    } else {
        info.insert("database".to_string(), "not_configured".to_string());
    }

    // Vérification de la connexion RabbitMQ
    if let Some(event_bus) = &state.event_bus {
        if event_bus.is_enabled {
            info.insert("rabbitmq".to_string(), "connected".to_string());
        } else {
            info.insert("rabbitmq".to_string(), "disabled".to_string());
        }
    } else {
        if state.config.rabbit_mq.enable {
            info.insert("rabbitmq".to_string(), "disconnected".to_string());
        } else {
            info.insert("rabbitmq".to_string(), "not_configured".to_string());
        }
    }

    Json(ApiResponse::success(info))
}

/// Récupération des messages
#[tracing::instrument(skip(state, params))]
async fn get_messages(
    State(state): State<AppState>,
    axum::extract::Path(conversation_id): axum::extract::Path<Uuid>, // Extract conversation_id from path
    Query(params): Query<GetMessagesQuery>,
) -> Result<Json<ApiResponse<Vec<Message>>>, StatusCode> {
    // Use Message model
    let limit = params.limit.unwrap_or(50).min(100);

    let messages = state
        .message_repo
        .get_conversation_messages(conversation_id, limit) // Use message_repo
        .await
        .map_err(|e| {
            warn!("Erreur récupération messages conversation: {}", e);
            StatusCode::INTERNAL_SERVER_ERROR
        })?;

    Ok(Json(ApiResponse::success(messages)))
}

/// Envoi de message
#[tracing::instrument(skip(state, payload))]
async fn send_message(
    State(state): State<AppState>,
    Json(payload): Json<SendMessageRequest>,
) -> Result<Json<ApiResponse<Uuid>>, StatusCode> {
    // Return Uuid
    let message = state
        .message_repo
        .create(payload.conversation_id, payload.sender_id, &payload.content) // Use message_repo
        .await
        .map_err(|e| {
            warn!("Erreur envoi message: {}", e);
            StatusCode::INTERNAL_SERVER_ERROR
        })?;

    info!(
        "✅ Message envoyé - ID: {:?}, sender: {:?}",
        message.id, message.sender_id
    );

    Ok(Json(ApiResponse::success(message.id)))
}

/// Statistiques basiques
#[tracing::instrument(skip(_state))]
async fn get_stats(State(_state): State<AppState>) -> Json<ApiResponse<HashMap<String, u32>>> {
    let mut stats = HashMap::new();
    stats.insert("total_messages".to_string(), 2);
    stats.insert("active_users".to_string(), 1);
    stats.insert("rooms".to_string(), 1);
    stats.insert("websocket_enabled".to_string(), 1);

    Json(ApiResponse::success(stats))
}