Mettre à jour WebSocketManager pour inclure l'initialisation, la gestion des tâches asynchrones, les statistiques de service et la collecte des données système, avec l'ajout de `psutil` pour surveiller l'utilisation de la mémoire. Ajuster les routes et la configuration pour améliorer la gestion des connexions WebSocket et des notifications.

config.yml

@@ -1,6 +1,7 @@
-APP_NAME: NotificationChannel
+APP_NAME: Socket service
 APP_ENV: local
-APP_PRODUCTION: False
+API_VERSION: 1.0.0
+APP_DEBUG: False
 APP_PORT: 7860
 APP_HOST: 0.0.0.0
 
@@ -10,3 +11,10 @@ WS_HOST: 0.0.0.0
 WS_PORT: 8765
 WS_SSL: false
 WS_CORS_ORIGINS: '*'
+
+WEBSOCKET_PING_INTERVAL: 20
+WEBSOCKET_TIMEOUT: 60
+
+CLEANUP_INTERVAL: 300
+HEARTBEAT_INTERVAL: 30
+STATS_INTERVAL: 60

core/manager/websocket_manager.py

@@ -1,6 +1,9 @@
+import asyncio
 from collections import deque
-from typing import Dict, List, Any, Deque
+from datetime import datetime, timezone
+from typing import Dict, List, Any, Deque, Optional
 
+import psutil
 from fastapi import WebSocket
 
 from core.settings.settings import Settings
@@ -20,8 +23,59 @@ class WebSocketManager:
             maxlen=self.settings.get("MAX_NOTIFICATION_HISTORY", 1000)
         )
 
+        self._start_time: Optional[datetime] = None
+        self._is_initialized: bool = False
+        self._shutdown_event: Optional[asyncio.Event] = None
+        self._background_tasks: List[asyncio.Task] = []
+
+        self._total_messages_sent: int = 0
+        self._total_connections_handled: int = 0
+        self._connection_errors: int = 0
+
+    async def initialize(self) -> None:
+        if self._is_initialized:
+            self.logger.warning("WebSocketManager already initialized")
+            return
+
+        self._start_time = datetime.now(timezone.utc)
+        self._shutdown_event = asyncio.Event()
+        self._is_initialized = True
+
+        self._background_tasks = [
+            asyncio.create_task(self._cleanup_expired_notifications()),
+            asyncio.create_task(self._heartbeat_checker()),
+            asyncio.create_task(self._stats_collector())
+        ]
+
+        self.logger.info("WebSocketManager initialized successfully")
+
+    async def shutdown(self) -> None:
+        if not self._is_initialized:
+            return
+
+        self.logger.info("Shutting down WebSocketManager...")
+
+        if self._shutdown_event:
+            self._shutdown_event.set()
+
+        for task in self._background_tasks:
+            if not task.done():
+                task.cancel()
+
+        if self._background_tasks:
+            await asyncio.gather(*self._background_tasks, return_exceptions=True)
+
+        await self._close_all_connections()
+
+        self.active_connections.clear()
+        self.client_info.clear()
+        self.subscriptions.clear()
+        self.notification_history.clear()
+
+        self._is_initialized = False
+        self.logger.info("WebSocketManager shutdown completed")
+
     async def connect(self, websocket: WebSocket, client: ClientInfo) -> str:
-        """Établit une nouvelle connexion WebSocket"""
         await websocket.accept()
         session_id = client.session_id
 
@@ -29,6 +83,8 @@ class WebSocketManager:
         self.client_info[session_id] = client
         self.subscriptions[session_id] = []
 
+        self._total_connections_handled += 1
+
         self.logger.info(f"Nouvelle connexion: {session_id}")
 
         await self._send(
@@ -45,7 +101,6 @@ class WebSocketManager:
         return session_id
 
     def disconnect(self, session_id: str) -> None:
-        """Ferme une connexion et nettoie les ressources"""
         if session_id in self.active_connections:
             del self.active_connections[session_id]
             del self.client_info[session_id]
@@ -55,21 +110,22 @@ class WebSocketManager:
     async def send_personal_message(
             self, session_id: str, payload: Dict[str, Any]
     ) -> None:
-        """Envoie un message à un client spécifique"""
         if ws := self.active_connections.get(session_id):
             try:
                 await self._send(ws, payload)
+                self._total_messages_sent += 1
             except RuntimeError as e:
                 self.logger.error(f"Erreur d'envoi à {session_id}: {e}")
+                self._connection_errors += 1
                 self.disconnect(session_id)
 
-    async def broadcast_notification(self, notification: Notification) -> None:
-        """Diffuse une notification aux clients abonnés"""
+    async def broadcast_notification(self, notification: Notification) -> Dict[str, Any]:
         if notification.is_expired:
-            return
+            return {"delivered_to": 0}
 
         self._add_to_history(notification)
         payload = self._format_notification(notification)
+        delivered_count = 0
 
         for session_id, subscriptions in self.subscriptions.items():
             if not (client := self.client_info.get(session_id)):
@@ -77,33 +133,41 @@ class WebSocketManager:
 
             if self._should_receive(subscriptions, notification, client):
                 await self.send_personal_message(session_id, payload)
+                delivered_count += 1
+
+        return {"delivered_to": delivered_count}
 
-    async def send_to_user(self, user_id: str, notification: Notification) -> None:
-        """Envoie une notification à un utilisateur spécifique"""
+    async def send_to_user(self, user_id: str, notification: Notification) -> Dict[str, Any]:
         if notification.is_expired:
-            return
+            return {"delivered_to": 0}
 
         self._add_to_history(notification)
         payload = self._format_notification(notification)
+        delivered_count = 0
 
         for session_id, client in self.client_info.items():
             if client.user_id == user_id:
                 await self.send_personal_message(session_id, payload)
+                delivered_count += 1
 
-    async def send_to_role(self, role: str, notification: Notification) -> None:
-        """Envoie une notification à un rôle spécifique"""
+        return {"delivered_to": delivered_count}
+
+    async def send_to_role(self, role: str, notification: Notification) -> Dict[str, Any]:
         if notification.is_expired:
-            return
+            return {"delivered_to": 0}
 
         self._add_to_history(notification)
         payload = self._format_notification(notification)
+        delivered_count = 0
 
         for session_id, client in self.client_info.items():
             if role in client.user_roles:
                 await self.send_personal_message(session_id, payload)
+                delivered_count += 1
+
+        return {"delivered_to": delivered_count}
 
     def add_subscription(self, session_id: str, subscription: Subscription) -> bool:
-        """Ajoute un nouvel abonnement pour un client"""
         if not (sub_list := self.subscriptions.get(session_id)):
             return False
 
@@ -115,7 +179,6 @@ class WebSocketManager:
         return True
 
     def update_subscription(self, session_id: str, subscription: Subscription) -> bool:
-        """Met à jour un abonnement existant"""
         if not (sub_list := self.subscriptions.get(session_id)):
             return False
 
@@ -126,7 +189,6 @@ class WebSocketManager:
         return False
 
     def remove_subscription(self, session_id: str, subscription_id: str) -> bool:
-        """Supprime un abonnement"""
         if not (sub_list := self.subscriptions.get(session_id)):
             return False
 
@@ -137,29 +199,146 @@ class WebSocketManager:
         return len(self.subscriptions[session_id]) < initial_count
 
     def get_subscriptions(self, session_id: str) -> List[Subscription]:
-        """Récupère les abonnements d'un client"""
         return self.subscriptions.get(session_id, [])[:]
 
-    def get_notification_history(self, limit: int = 100) -> List[Notification]:
-        """Récupère l'historique des notifications"""
-        return [n for n in list(self.notification_history)[-limit:] if not n.is_expired]
+    def get_notification_history(self, limit: int = 100, offset: int = 0) -> List[Notification]:
+        history = [n for n in list(self.notification_history) if not n.is_expired]
+        return history[offset:offset + limit]
 
     def get_stats(self) -> Dict[str, Any]:
-        """Retourne les statistiques du service"""
+        memory_info = self.get_memory_usage()
+
         return {
             "active_connections": len(self.active_connections),
             "total_subscriptions": sum(len(s) for s in self.subscriptions.values()),
             "total_notifications": len(self.notification_history),
+            "total_messages_sent": self._total_messages_sent,
+            "total_connections_handled": self._total_connections_handled,
+            "connection_errors": self._connection_errors,
+            "uptime_seconds": self.get_uptime(),
+            "memory_usage": memory_info,
             "timestamp": current_utc_iso(),
         }
 
     @staticmethod
-    async def _send(websocket: WebSocket, payload: Dict[str, Any]) -> None:
-        """Mécanisme d'envoi générique"""
-        await websocket.send_json(payload)
+    def get_current_timestamp() -> str:
+        return current_utc_iso()
+
+    def get_uptime(self) -> float:
+        if not self._start_time:
+            return 0.0
+        return (datetime.now(timezone.utc) - self._start_time).total_seconds()
+
+    def get_memory_usage(self) -> Dict[str, Any]:
+        try:
+            process = psutil.Process()
+            memory_info = process.memory_info()
+
+            return {
+                "rss_mb": round(memory_info.rss / 1024 / 1024, 2),  # Resident Set Size
+                "vms_mb": round(memory_info.vms / 1024 / 1024, 2),  # Virtual Memory Size
+                "percent": round(process.memory_percent(), 2),
+                "available_mb": round(psutil.virtual_memory().available / 1024 / 1024, 2),
+                "total_mb": round(psutil.virtual_memory().total / 1024 / 1024, 2)
+            }
+        except Exception as e:
+            self.logger.error(f"Error getting memory usage: {e}")
+            return {
+                "rss_mb": 0,
+                "vms_mb": 0,
+                "percent": 0,
+                "available_mb": 0,
+                "total_mb": 0,
+                "error": str(e)
+            }
+
+    async def _cleanup_expired_notifications(self) -> None:
+        while not self._shutdown_event.is_set():
+            try:
+                active_notifications = [
+                    n for n in self.notification_history if not n.is_expired
+                ]
+
+                self.notification_history.clear()
+                self.notification_history.extend(active_notifications)
+
+                await asyncio.sleep(self.settings.get("CLEANUP_INTERVAL", 300))  # 5 minutes
+
+            except asyncio.CancelledError:
+                break
+            except Exception as e:
+                self.logger.error(f"Error in cleanup task: {e}")
+                await asyncio.sleep(60)
+
+    async def _heartbeat_checker(self) -> None:
+        while not self._shutdown_event.is_set():
+            try:
+                disconnected_sessions = []
+
+                for session_id, websocket in self.active_connections.items():
+                    try:
+                        await websocket.send_json({"type": "ping", "timestamp": current_utc_iso()})
+                    except Exception as e:
+                        self.logger.warning(f"Connection {session_id} appears to be dead: {e}")
+                        disconnected_sessions.append(session_id)
+
+                for session_id in disconnected_sessions:
+                    self.disconnect(session_id)
+
+                await asyncio.sleep(self.settings.get("HEARTBEAT_INTERVAL", 30))  # 30 seconds
+
+            except asyncio.CancelledError:
+                break
+            except Exception as e:
+                self.logger.error(f"Error in heartbeat checker: {e}")
+                await asyncio.sleep(60)
+
+    async def _stats_collector(self) -> None:
+        while not self._shutdown_event.is_set():
+            try:
+                stats = self.get_stats()
+                self.logger.info(f"Service stats: {stats}")
+
+                await asyncio.sleep(self.settings.get("STATS_INTERVAL", 60))  # 1 minute
+
+            except asyncio.CancelledError:
+                break
+            except Exception as e:
+                self.logger.error(f"Error in stats collector: {e}")
+                await asyncio.sleep(60)
+
+    async def _close_all_connections(self) -> None:
+        if not self.active_connections:
+            return
+
+        self.logger.info(f"Closing {len(self.active_connections)} active connections...")
+
+        shutdown_message = {
+            "type": "service_shutdown",
+            "title": "Service en cours d'arrêt",
+            "message": "Le service de notifications va être arrêté",
+            "timestamp": current_utc_iso()
+        }
+
+        close_tasks = []
+        for session_id, websocket in self.active_connections.items():
+            try:
+                await self._send(websocket, shutdown_message)
+                close_tasks.append(websocket.close())
+            except Exception as e:
+                self.logger.error(f"Error closing connection {session_id}: {e}")
+
+        if close_tasks:
+            await asyncio.gather(*close_tasks, return_exceptions=True)
+
+    async def _send(self, websocket: WebSocket, payload: Dict[str, Any]) -> None:
+        try:
+            await websocket.send_json(payload)
+        except Exception as e:
+            self.logger.error(f"Failed to send message: {e}")
+            raise RuntimeError(f"WebSocket send failed: {e}")
 
     def _add_to_history(self, notification: Notification) -> None:
-        """Ajoute une notification à l'historique"""
         if not notification.is_expired:
             self.notification_history.append(notification)
 
@@ -169,7 +348,6 @@ class WebSocketManager:
             notification: Notification,
             client: ClientInfo,
     ) -> bool:
-        """Vérifie si un client doit recevoir une notification"""
         return any(
             sub.matches(notification, client.user_id, client.user_roles)
             for sub in subscriptions
@@ -177,5 +355,4 @@ class WebSocketManager:
 
     @staticmethod
     def _format_notification(notification: Notification) -> Dict[str, Any]:
-        """Formate une notification pour l'envoi"""
         return {"type": "notification", **notification.model_dump(exclude_unset=True)}

core/types/datamodels.py

@@ -63,7 +63,7 @@ class Subscription(BaseModel):
     created_at: datetime = Field(default_factory=datetime.now)
 
     def matches(
-        self, notification: "Notification", user_id: str, user_roles: List[str]
+            self, notification: "Notification", user_id: str, user_roles: List[str]
     ) -> bool:
         if self.notification_types and notification.type not in self.notification_types:
             return False
@@ -83,3 +83,5 @@ class ClientInfo(BaseModel):
     user_id: str
     user_roles: List[str] = Field(default_factory=list)
     connected_at: datetime = Field(default_factory=datetime.now)
+    ip_address: str
+    user_agent: str

main.py

@@ -1,19 +1,10 @@
 from contextlib import asynccontextmanager
-from pathlib import Path
-from typing import Any, Dict, List, Optional
+from typing import List, Optional
 from uuid import uuid4
 
 import uvicorn
-from fastapi import (
-    FastAPI,
-    WebSocket,
-    Request,
-    WebSocketDisconnect,
-    HTTPException,
-    Depends,
-)
-from fastapi.templating import Jinja2Templates
-from starlette.responses import HTMLResponse
+from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException, Depends, status
+from fastapi.middleware.cors import CORSMiddleware
 
 from core.manager.websocket_manager import WebSocketManager
 from core.settings.settings import Settings
@@ -25,220 +16,397 @@ from core.types.datamodels import (
     Notification,
     BroadcastRequest,
 )
-from core.utils.utils import current_utc_iso
-
-BASE_DIR = Path(__file__).resolve().parent.absolute()
-templates = Jinja2Templates(directory=str(BASE_DIR / "templates"))
-
-settings = Settings()
-manager = WebSocketManager(settings=settings)
 
-logger = manager.logger
 
-
-async def validate_session(session_id: str) -> str:
-    if session_id not in manager.active_connections:
-        raise HTTPException(
-            status_code=404, detail="Session introuvable ou déjà fermée"
+class WebSocketNotificationService:
+    def __init__(self, settings: Optional[Settings] = None):
+        self.settings = settings or Settings()
+        self.manager = WebSocketManager(settings=self.settings)
+        self.logger = self.manager.logger
+        self.app = self._create_app()
+        self._configure_middleware()
+        self._register_routes()
+
+    def _create_app(self) -> FastAPI:
+        return FastAPI(
+            title=self.settings.APP_NAME,
+            description="Serivce de notification WebSocket et de real-time data",
+            version=self.settings.API_VERSION,
+            lifespan=self._lifespan,
+            docs_url="/docs" if self.settings.APP_DEBUG else None,
+            redoc_url="/redoc" if self.settings.APP_DEBUG else None,
+            websocket_ping_interval=self.settings.WEBSOCKET_PING_INTERVAL,
+            websocket_timeout=self.settings.WEBSOCKET_TIMEOUT,
+            openapi_tags=[
+                {
+                    "name": "health",
+                    "description": "Santé et vérification du système"
+                },
+                {
+                    "name": "websocket",
+                    "description": "Gestion de connexions WebSocket"
+                },
+                {
+                    "name": "subscriptions",
+                    "description": "Gestion des subscriptions"
+                },
+                {
+                    "name": "notifications",
+                    "description": "Envoie et gestion des notifications"
+                },
+                {
+                    "name": "monitoring",
+                    "description": "Analytics et statistiques"
+                }
+            ]
         )
-    return session_id
-
-
-@asynccontextmanager
-async def lifespan(_app: FastAPI):
-    """Gère les événements de démarrage et d'arrêt de l'application"""
-
-    logger.info("Démarrage du service de notifications WebSocket")
-
-    yield
-
-    logger.info("Arrêt du service de notifications WebSocket")
 
-
-app = FastAPI(
-    title=manager.settings.APP_NAME,
-    description="Service de notifications en temps réel avec support WebSocket",
-    version="1.0.0",
-    lifespan=lifespan,
-    docs_url="/docs",
-    redoc_url="/redoc",
-    websocket_ping_interval=20,
-    websocket_timeout=60,
-)
-
-
-@app.websocket("/ws/{user_id}")
-async def websocket_endpoint(websocket: WebSocket, user_id: str):
-    """Endpoint pour les connexions WebSocket"""
-    session_id = ""
-
-    try:
-        client_info = ClientInfo(
-            session_id=str(uuid4()),
-            user_id=user_id,
-            user_roles=(
-                websocket.query_params.get("roles", "").split(",")
-                if websocket.query_params.get("roles")
-                else []
-            ),
+    def _configure_middleware(self):
+        self.app.add_middleware(
+            CORSMiddleware,
+            allow_origins=[self.settings.WS_CORS_ORIGINS],
+            allow_credentials=True,
+            allow_methods=["*"],
+            allow_headers=["*"],
         )
 
-        session_id = await manager.connect(websocket, client_info)
-
+    @asynccontextmanager
+    async def _lifespan(self, app: FastAPI):
+        self.logger.info("🚀 Starting WebSocket notification service")
+
+        try:
+            await self.manager.initialize()
+            yield
+        finally:
+            await self.manager.shutdown()
+            self.logger.info("🛑 WebSocket notification service stopped")
+
+    def _register_routes(self):
+        self._register_health_routes()
+        self._register_websocket_routes()
+        self._register_subscription_routes()
+        self._register_notification_routes()
+        self._register_monitoring_routes()
+
+    def _register_health_routes(self):
+        @self.app.get("/", tags=["health"])
+        async def root():
+            return {
+                "service": self.settings.APP_NAME,
+                "version": self.settings.API_VERSION,
+                "status": "healthy",
+                "timestamp": self.manager.get_current_timestamp()
+            }
+
+        @self.app.get("/health", tags=["health"])
+        async def health_check():
+            return {
+                "status": "healthy",
+                "connections": len(self.manager.active_connections),
+                "uptime": self.manager.get_uptime(),
+                "memory_usage": self.manager.get_memory_usage()
+            }
+
+    def _register_websocket_routes(self):
+        @self.app.websocket("/ws/{user_id}")
+        async def websocket_endpoint(websocket: WebSocket, user_id: str):
+            session_id = ""
+
+            try:
+                client_info = ClientInfo(
+                    session_id=str(uuid4()),
+                    user_id=user_id,
+                    user_roles=self._extract_roles(websocket),
+                    ip_address=self._get_client_ip(websocket),
+                    user_agent=self._get_user_agent(websocket)
+                )
+
+                session_id = await self.manager.connect(websocket, client_info)
+
+                await self._handle_websocket_messages(websocket, session_id)
+
+            except WebSocketDisconnect:
+                self.logger.info(f"WebSocket connection closed: {session_id}")
+            except Exception as e:
+                self.logger.error(f"WebSocket error: {e}")
+                await self._handle_websocket_error(websocket, e)
+            finally:
+                self.manager.disconnect(session_id)
+
+    def _register_subscription_routes(self):
+        @self.app.post("/subscriptions/{session_id}", tags=["subscriptions"])
+        async def add_subscription(
+                subscription_request: SubscriptionRequest,
+                session_id: str = Depends(self._validate_session),
+        ):
+            try:
+                subscription = Subscription(**subscription_request.model_dump())
+
+                if not self.manager.add_subscription(session_id, subscription):
+                    raise HTTPException(
+                        status_code=status.HTTP_404_NOT_FOUND,
+                        detail="Failed to add subscription"
+                    )
+
+                return {
+                    "success": True,
+                    "message": "Subscription added successfully",
+                    "subscription_id": subscription.id,
+                    "timestamp": self.manager.get_current_timestamp()
+                }
+            except Exception as e:
+                raise HTTPException(
+                    status_code=status.HTTP_400_BAD_REQUEST,
+                    detail=str(e)
+                )
+
+        @self.app.delete("/subscriptions/{session_id}/{subscription_id}", tags=["subscriptions"])
+        async def remove_subscription(
+                subscription_id: str,
+                session_id: str = Depends(self._validate_session),
+        ):
+            if not self.manager.remove_subscription(session_id, subscription_id):
+                raise HTTPException(
+                    status_code=status.HTTP_404_NOT_FOUND,
+                    detail="Subscription not found"
+                )
+
+            return {
+                "success": True,
+                "message": "Subscription removed successfully",
+                "timestamp": self.manager.get_current_timestamp()
+            }
+
+        @self.app.get("/subscriptions/{session_id}", tags=["subscriptions"])
+        async def get_subscriptions(
+                session_id: str = Depends(self._validate_session),
+        ):
+            """Get all subscriptions for a session"""
+            subscriptions = self.manager.get_subscriptions(session_id)
+            return {
+                "success": True,
+                "subscriptions": subscriptions,
+                "count": len(subscriptions),
+                "timestamp": self.manager.get_current_timestamp()
+            }
+
+    def _register_notification_routes(self):
+        @self.app.post("/notifications/user/{user_id}", tags=["notifications"])
+        async def send_user_notification(
+                user_id: str,
+                notification_request: NotificationRequest,
+        ):
+            try:
+                notification = Notification(**notification_request.model_dump())
+                result = await self.manager.send_to_user(user_id, notification)
+
+                return {
+                    "success": True,
+                    "message": f"Notification sent to user {user_id}",
+                    "notification_id": notification.id,
+                    "delivered_to": result.get("delivered_to", 0),
+                    "timestamp": self.manager.get_current_timestamp()
+                }
+            except Exception as e:
+                raise HTTPException(
+                    status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                    detail=str(e)
+                )
+
+        @self.app.post("/notifications/role/{role}", tags=["notifications"])
+        async def send_role_notification(
+                role: str,
+                notification_request: NotificationRequest,
+        ):
+            try:
+                notification = Notification(**notification_request.model_dump())
+                result = await self.manager.send_to_role(role, notification)
+
+                return {
+                    "success": True,
+                    "message": f"Notification sent to role {role}",
+                    "notification_id": notification.id,
+                    "delivered_to": result.get("delivered_to", 0),
+                    "timestamp": self.manager.get_current_timestamp()
+                }
+            except Exception as e:
+                raise HTTPException(
+                    status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                    detail=str(e)
+                )
+
+        @self.app.post("/notifications/broadcast", tags=["notifications"])
+        async def broadcast_notification(
+                notification_request: NotificationRequest,
+        ):
+            try:
+                notification = Notification(**notification_request.model_dump())
+                result = await self.manager.broadcast_notification(notification)
+
+                return {
+                    "success": True,
+                    "message": "Notification broadcasted",
+                    "notification_id": notification.id,
+                    "delivered_to": result.get("delivered_to", 0),
+                    "timestamp": self.manager.get_current_timestamp()
+                }
+            except Exception as e:
+                raise HTTPException(
+                    status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                    detail=str(e)
+                )
+
+        @self.app.post("/notifications/send", tags=["notifications"])
+        async def send_targeted_notification(
+                request: BroadcastRequest,
+        ):
+            try:
+                notification = Notification(**request.notification.model_dump())
+
+                if request.target_type == "user":
+                    result = await self.manager.send_to_user(request.target_value, notification)
+                elif request.target_type == "role":
+                    result = await self.manager.send_to_role(request.target_value, notification)
+                elif request.target_type == "global":
+                    result = await self.manager.broadcast_notification(notification)
+                else:
+                    raise ValueError("Invalid target type")
+
+                return {
+                    "success": True,
+                    "message": f"Notification sent to {request.target_type}: {request.target_value}",
+                    "notification_id": notification.id,
+                    "delivered_to": result.get("delivered_to", 0),
+                    "timestamp": self.manager.get_current_timestamp()
+                }
+            except Exception as e:
+                raise HTTPException(
+                    status_code=status.HTTP_400_BAD_REQUEST,
+                    detail=str(e)
+                )
+
+    def _register_monitoring_routes(self):
+        @self.app.get("/notifications/history", tags=["monitoring"])
+        async def get_notification_history(
+                limit: int = 100,
+                offset: int = 0,
+        ):
+            history = self.manager.get_notification_history(limit, offset)
+            return {
+                "success": True,
+                "history": history,
+                "count": len(history),
+                "limit": limit,
+                "offset": offset,
+                "timestamp": self.manager.get_current_timestamp()
+            }
+
+        @self.app.get("/stats", tags=["monitoring"])
+        async def get_service_stats():
+            stats = self.manager.get_stats()
+            return {
+                "success": True,
+                "stats": stats,
+                "timestamp": self.manager.get_current_timestamp()
+            }
+
+        @self.app.get("/clients", tags=["monitoring"])
+        async def get_connected_clients():
+            clients = [
+                {
+                    "session_id": session_id,
+                    "user_id": info.user_id,
+                    "user_roles": info.user_roles,
+                    "ip_address": info.ip_address,
+                    "connected_at": info.connected_at.isoformat(),
+                    "subscriptions_count": len(self.manager.subscriptions.get(session_id, [])),
+                    "last_activity": info.last_activity.isoformat() if hasattr(info, 'last_activity') else None
+                }
+                for session_id, info in self.manager.client_info.items()
+            ]
+
+            return {
+                "success": True,
+                "clients": clients,
+                "total_connections": len(clients),
+                "timestamp": self.manager.get_current_timestamp()
+            }
+
+    # Helper methods
+    async def _validate_session(self, session_id: str) -> str:
+        if session_id not in self.manager.active_connections:
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail="Session not found or already closed"
+            )
+        return session_id
+
+    @staticmethod
+    def _extract_roles(websocket: WebSocket) -> List[str]:
+        roles_param = websocket.query_params.get("roles", "")
+        return roles_param.split(",") if roles_param else []
+
+    @staticmethod
+    def _get_client_ip(websocket: WebSocket) -> str:
+        return websocket.client.host if websocket.client else "unknown"
+
+    @staticmethod
+    def _get_user_agent(websocket: WebSocket) -> str:
+        return websocket.headers.get("user-agent", "unknown")
+
+    async def _handle_websocket_messages(self, websocket: WebSocket, session_id: str):
         while True:
-            message = await websocket.receive_text()
-            if message == "ping":
-                await websocket.send_text("pong")
-            else:
-                pass
-
-    except WebSocketDisconnect:
-        manager.disconnect(session_id)
-        logger.info(f"Connexion WebSocket fermée: {session_id}")
-    except Exception as e:
-        logger.error(f"Erreur WebSocket: {e}")
-        manager.disconnect(session_id)
-
-
-# Endpoints REST
-@app.get("/", response_class=HTMLResponse)
-async def root(request: Request):
-    """Page de test pour les connexions WebSocket"""
-    return templates.TemplateResponse(
-        "dashboard.html",
-        {"request": request, "current_time": current_utc_iso()},
-    )
-
-
-@app.post("/subscriptions/{session_id}")
-async def add_subscription(
-        session_id: str = Depends(validate_session),
-        subscription_request: Optional[SubscriptionRequest] = None,
-):
-    """Ajoute un nouvel abonnement pour une session"""
-
-    subscription = Subscription(**subscription_request.model_dump()) if subscription_request else Subscription(
-        type="default", target="default")
-
-    if not manager.add_subscription(session_id, subscription):
-        raise HTTPException(status_code=400, detail="Échec de l'ajout de l'abonnement")
-
-    return {
-        "message": "Abonnement ajouté avec succès",
-        "subscription_id": subscription.id,
-    }
-
-
-@app.delete("/subscriptions/{session_id}/{subscription_id}")
-async def remove_subscription(
-        subscription_id: str,
-        session_id: str = Depends(validate_session),
-):
-    """Supprime un abonnement existant"""
-    if not manager.remove_subscription(session_id, subscription_id):
-        raise HTTPException(status_code=404, detail="Abonnement introuvable")
-
-    return {"message": "Abonnement supprimé avec succès"}
-
-
-@app.get("/subscriptions/{session_id}")
-async def get_subscriptions(session_id: str = Depends(validate_session)):
-    """Récupère les abonnements d'une session"""
-    return manager.get_subscriptions(session_id)
-
-
-@app.post("/notifications/user/{user_id}")
-async def send_user_notification(
-        user_id: str, notification_request: NotificationRequest
-):
-    """Envoie une notification à un utilisateur spécifique"""
-    notification = Notification(**notification_request.model_dump())
-    await manager.send_to_user(user_id, notification)
-
-    return {
-        "message": f"Notification envoyée à l'utilisateur {user_id}",
-        "notification_id": notification.id,
-    }
-
-
-@app.post("/notifications/role/{role}")
-async def send_role_notification(role: str, notification_request: NotificationRequest):
-    """Envoie une notification à un rôle spécifique"""
-    notification = Notification(**notification_request.model_dump())
-    await manager.send_to_role(role, notification)
-
-    return {
-        "message": f"Notification envoyée au rôle {role}",
-        "notification_id": notification.id,
-    }
-
-
-@app.post("/notifications/broadcast")
-async def broadcast_notification(notification_request: NotificationRequest):
-    """Diffuse une notification à tous les clients connectés"""
-    notification = Notification(**notification_request.model_dump())
-    await manager.broadcast_notification(notification)
-
-    return {
-        "message": "Notification diffusée",
-        "notification_id": notification.id,
-    }
-
-
-@app.post("/notifications/send")
-async def send_targeted_notification(request: BroadcastRequest):
-    """Envoie une notification ciblée"""
-    notification = Notification(**request.notification.model_dump())
-
-    if request.target_type == "user":
-        await manager.send_to_user(request.target_value, notification)
-    elif request.target_type == "role":
-        await manager.send_to_role(request.target_value, notification)
-    elif request.target_type == "global":
-        await manager.broadcast_notification(notification)
-    else:
-        raise HTTPException(status_code=400, detail="Type de cible invalide")
-
-    return {
-        "message": f"Notification envoyée à {request.target_type}: {request.target_value}",
-        "notification_id": notification.id,
-    }
-
-
-@app.get("/notifications/history")
-async def get_notification_history(limit: int = 100):
-    """Récupère l'historique des notifications"""
-    return manager.get_notification_history(limit)
-
-
-@app.get("/stats")
-async def get_service_stats():
-    """Récupère les statistiques du service"""
-    return manager.get_stats()
+            try:
+                message = await websocket.receive_text()
+                await self._process_websocket_message(websocket, session_id, message)
+            except WebSocketDisconnect:
+                raise
+            except Exception as e:
+                self.logger.error(f"Error processing WebSocket message: {e}")
+                await websocket.send_text(f"Error: {str(e)}")
+
+    async def _process_websocket_message(self, websocket: WebSocket, session_id: str, message: str):
+        if message == "ping":
+            await websocket.send_text("pong")
+        elif message.startswith("subscribe:"):
+            topic = message.replace("subscribe:", "")
+            subscription = Subscription(type="topic", target=topic)
+            self.manager.add_subscription(session_id, subscription)
+            await websocket.send_text(f"subscribed:{topic}")
+        elif message.startswith("unsubscribe:"):
+            topic = message.replace("unsubscribe:", "")
+            await websocket.send_text(f"unsubscribed:{topic}")
+        else:
+            await self._handle_custom_message(websocket, session_id, message)
+
+    async def _handle_custom_message(self, websocket: WebSocket, session_id: str, message: str):
+        """Handle custom WebSocket messages"""
+        pass
+
+    @staticmethod
+    async def _handle_websocket_error(websocket: WebSocket, error: Exception):
+        """Handle WebSocket errors gracefully"""
+        try:
+            await websocket.send_text(f"Error: {str(error)}")
+        except:
+            raise
+
+    def run(self):
+        uvicorn.run(
+            self.app,
+            host=self.settings.APP_HOST,
+            port=self.settings.APP_PORT,
+            reload=self.settings.APP_DEBUG,
+            log_level="info",
+            server_header=False,
+            access_log=self.settings.APP_DEBUG
+        )
 
 
-@app.get("/clients")
-async def get_connected_clients():
-    """Récupère la liste des clients connectés"""
-    clients: List[Dict[str, Any]] = [
-        {
-            "session_id": session_id,
-            "user_id": info.user_id,
-            "user_roles": info.user_roles,
-            "connected_at": info.connected_at.isoformat(),
-            "subscriptions": len(manager.subscriptions.get(session_id, [])),
-        }
-        for session_id, info in manager.client_info.items()
-    ]
-    return {"clients": clients}
+def main():
+    service = WebSocketNotificationService()
+    service.run()
 
 
 if __name__ == "__main__":
-    uvicorn.run(
-        "main:app",
-        host=manager.settings.APP_HOST,
-        port=manager.settings.APP_PORT,
-        reload=True,
-        log_level="info",
-        server_header=False,
-    )
+    main()

requirements.txt

@@ -3,4 +3,5 @@ PyYAML
 pydantic
 fastapi
 uvicorn[standard]
-jinja2
+jinja2
+psutil