# backend/modules/fleet/services/pusher_service.py """ Pusher Integration Service for Real-time Notifications Handles broadcasting notifications to users via WebSocket channels """ import logging import json from typing import Optional, Dict, Any from django.conf import settings from django.core.serializers.json import DjangoJSONEncoder try: import pusher PUSHER_AVAILABLE = True except ImportError: PUSHER_AVAILABLE = False pusher = None logger = logging.getLogger(__name__) class PusherService: """ Wrapper service for Pusher integration. Handles event broadcasting, channel authentication, and error handling. """ _instance = None _client = None def __new__(cls): """Singleton pattern to reuse Pusher client""" if cls._instance is None: cls._instance = super(PusherService, cls).__new__(cls) return cls._instance @classmethod def get_client(cls): """Get or initialize Pusher client""" if cls._client is None: if not PUSHER_AVAILABLE: logger.warning("Pusher library not installed. Install with: pip install pusher") return None try: pusher_config = { 'app_id': settings.PUSHER_APP_ID, 'key': settings.PUSHER_KEY, 'secret': settings.PUSHER_SECRET, 'cluster': settings.PUSHER_CLUSTER, 'ssl': True, 'json_encoder': DjangoJSONEncoder, } cls._client = pusher.Pusher(**pusher_config) except AttributeError as e: logger.error(f"Pusher configuration missing: {e}") return None return cls._client @staticmethod def is_available() -> bool: """Check if Pusher is available and configured""" if not PUSHER_AVAILABLE: return False required_values = [ settings.PUSHER_APP_ID, settings.PUSHER_KEY, settings.PUSHER_SECRET, settings.PUSHER_CLUSTER, ] return all(required_values) @staticmethod def get_private_channel_name(user_id: int) -> str: """Generate private channel name for user""" return f"private-user-{user_id}" @staticmethod def get_presence_channel_name() -> str: """Get the presence channel name for fleet activity""" return "presence-fleet" @classmethod def broadcast_notification( cls, user_id: int, title: str, message: str, level: str = 'info', notification_id: Optional[Any] = None, metadata: Optional[Dict[str, Any]] = None ) -> bool: """ Broadcast a notification to a specific user via private channel. Args: user_id: ID of the user to notify title: Notification title message: Notification message level: Notification level (info, warning, critical) notification_id: ID of the VehicleNotification record metadata: Additional metadata to send with notification Returns: True if broadcast successful, False otherwise """ if not cls.is_available(): logger.debug("Pusher not available, skipping broadcast") return False client = cls.get_client() if not client: return False channel_name = cls.get_private_channel_name(user_id) event_name = "vehicle.notification.created" from django.utils.timezone import now event_data = { 'id': cls._to_json_safe(notification_id), 'title': title, 'message': message, 'level': level, 'is_read': False, 'created_at': now().isoformat(), } if metadata: event_data['metadata'] = cls._to_json_safe(metadata) try: response = client.trigger( channels=channel_name, event_name=event_name, data=event_data ) logger.info(f"Notification broadcasted to {channel_name}: {event_name}") return True except Exception as e: logger.exception(f"Error broadcasting notification to {channel_name}: {e}") return False @staticmethod def _to_json_safe(value: Any) -> Any: """ Convert complex Python/Django values (UUID, date, Decimal, etc.) into JSON-serializable primitives using DjangoJSONEncoder. """ return json.loads(json.dumps(value, cls=DjangoJSONEncoder)) @classmethod def broadcast_to_multiple_users( cls, user_ids: list, title: str, message: str, level: str = 'info', notification_ids: Optional[Dict[int, int]] = None ) -> Dict[int, bool]: """ Broadcast the same notification to multiple users. Args: user_ids: List of user IDs to notify title: Notification title message: Notification message level: Notification level notification_ids: Dict mapping user_id to notification_id Returns: Dict of user_id -> success boolean """ results = {} for user_id in user_ids: notification_id = notification_ids.get(user_id) if notification_ids else None results[user_id] = cls.broadcast_notification( user_id=user_id, title=title, message=message, level=level, notification_id=notification_id ) return results @classmethod def authenticate_channel(cls, socket_id: str, channel_name: str, user_id: int) -> Optional[Dict[str, Any]]: """ Authenticate a user for a private or presence channel. Args: socket_id: Pusher socket ID from client channel_name: Channel name to authenticate user_id: Current user ID Returns: Signed authentication data or None if authentication fails """ if not cls.is_available(): logger.debug("Pusher not available for authentication") return None client = cls.get_client() if not client: return None # Only allow subscription to user's own private channel expected_channel = cls.get_private_channel_name(user_id) if channel_name != expected_channel and channel_name != cls.get_presence_channel_name(): logger.warning(f"Unauthorized channel subscription attempt: {channel_name} by user {user_id}") return None try: auth_data = client.authenticate(channel=channel_name, socket_id=socket_id) logger.info(f"Channel {channel_name} authenticated for user {user_id}") return auth_data except Exception as e: logger.exception(f"Error authenticating channel {channel_name}: {e}") return None