477 lines
17 KiB
Rust
477 lines
17 KiB
Rust
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)
|
|
.with_graceful_shutdown(shutdown_signal())
|
|
.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))
|
|
}
|
|
|
|
/// Gestionnaire de signal d'arrêt (Graceful Shutdown)
|
|
async fn shutdown_signal() {
|
|
let ctrl_c = async {
|
|
tokio::signal::ctrl_c()
|
|
.await
|
|
.expect("failed to install Ctrl+C handler");
|
|
};
|
|
|
|
#[cfg(unix)]
|
|
let terminate = async {
|
|
tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
|
|
.expect("failed to install signal handler")
|
|
.recv()
|
|
.await;
|
|
};
|
|
|
|
#[cfg(not(unix))]
|
|
let terminate = std::future::pending::<()>();
|
|
|
|
tokio::select! {
|
|
_ = ctrl_c => {},
|
|
_ = terminate => {},
|
|
}
|
|
|
|
info!("🛑 Signal d'arrêt reçu, fermeture gracieuse...");
|
|
}
|