""" Push Notification Service for RateRight PWA Handles push notifications via Web Push Protocol using PyWebPush """ import json import logging from datetime import datetime from typi, timezoneng import Dict, List, Optional, Any from pywebpush import webpush, WebPushException from app.extensions import db from app.models.notification import Notification from sqlalchemy import Column, Integer, String, Text, DateTime, Boolean from sqlalchemy.ext.declarative import declarative_base logger = logging.getLogger(__name__) # Database model for push subscriptions class PushSubscription(db.Model): __tablename__ = 'push_subscriptions' id = Column(Integer, primary_key=True) user_id = Column(Integer, db.ForeignKey('users.id'), nullable=False) endpoint = Column(Text, nullable=False) p256dh_key = Column(String(255), nullable=False) auth_key = Column(String(255), nullable=False) created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) is_active = Column(Boolean, default=True) def to_dict(self): return { 'endpoint': self.endpoint, 'keys': { 'p256dh': self.p256dh_key, 'auth': self.auth_key } } class PushNotificationService: """Service for handling push notifications""" def __init__(self): # You need to generate VAPID keys for production # vapid_gen.Vapid().generate_keys() self.vapid_private_key = self._get_vapid_private_key() self.vapid_public_key = self._get_vapid_public_key() self.vapid_claims = { "sub": "mailto:support@rateright.com.au" } def _get_vapid_private_key(self) -> str: """Get VAPID private key from environment or config""" # In production, store this securely in environment variables return """-----BEGIN EC PRIVATE KEY----- MHcCAQEEIBKKyGQgbIrwjXRBUHdJjHvFc6CuI/d7DXdqb8xj5KmPoAoGCCqGSM49 AwEHoUQDQgAE7wI9Df/2VAEyoU8/jG/M4LZY8YWyGAOGwJgZCe9ZRgLdJgRhBdLZ 0Gf4CqR7yPsH8jRNzBwB5GJHhBGXy7e9Zw== -----END EC PRIVATE KEY-----""" def _get_vapid_public_key(self) -> str: """Get VAPID public key""" return "BCPrMjq6Pt13KKPR-FFKjqfVQjcHyKaP6_D9dFkUhZp2_wYJqUGFQ7C8Yb9zKx4UmGFJhD6sQJj6vF_3wR9cO-E" def subscribe_user(self, user_id: int, subscription_data: Dict) -> bool: """Subscribe user to push notifications""" try: # Parse subscription data endpoint = subscription_data.get('endpoint') keys = subscription_data.get('keys', {}) p256dh = keys.get('p256dh') auth = keys.get('auth') if not all([endpoint, p256dh, auth]): logger.error("Invalid subscription data") return False # Check if subscription already exists existing = PushSubscription.query.filter_by( user_id=user_id, endpoint=endpoint ).first() if existing: # Update existing subscription existing.p256dh_key = p256dh existing.auth_key = auth existing.is_active = True existing.updated_at = datetime.now(timezone.utc) else: # Create new subscription subscription = PushSubscription( user_id=user_id, endpoint=endpoint, p256dh_key=p256dh, auth_key=auth ) db.session.add(subscription) db.session.commit() logger.info(f"Push subscription created/updated for user {user_id}") return True except Exception as e: logger.error(f"Failed to subscribe user {user_id}: {str(e)}") db.session.rollback() return False def unsubscribe_user(self, user_id: int, endpoint: str = None) -> bool: """Unsubscribe user from push notifications""" try: query = PushSubscription.query.filter_by(user_id=user_id) if endpoint: query = query.filter_by(endpoint=endpoint) subscriptions = query.all() for subscription in subscriptions: subscription.is_active = False db.session.commit() logger.info(f"Push subscriptions deactivated for user {user_id}") return True except Exception as e: logger.error(f"Failed to unsubscribe user {user_id}: {str(e)}") return False def send_notification_to_user(self, user_id: int, title: str, body: str, data: Optional[Dict] = None, actions: Optional[List[Dict]] = None) -> bool: """Send push notification to a specific user""" try: subscriptions = PushSubscription.query.filter_by( user_id=user_id, is_active=True ).all() if not subscriptions: logger.info(f"No active push subscriptions for user {user_id}") return False notification_data = { 'title': title, 'body': body, 'icon': '/static/icons/icon-192x192.png', 'badge': '/static/icons/badge-72x72.png', 'tag': f'rateright-{user_id}-{datetime.now().timestamp()}', 'requireInteraction': False, 'data': data or {}, 'actions': actions or [ { 'action': 'view', 'title': 'View', 'icon': '/static/icons/action-view.png' }, { 'action': 'dismiss', 'title': 'Dismiss', 'icon': '/static/icons/action-dismiss.png' } ] } successful_sends = 0 failed_subscriptions = [] for subscription in subscriptions: try: response = webpush( subscription_info=subscription.to_dict(), data=json.dumps(notification_data), vapid_private_key=self.vapid_private_key, vapid_claims=self.vapid_claims, content_encoding='aes128gcm' ) if response.status_code in [200, 201, 204]: successful_sends += 1 logger.debug(f"Push notification sent successfully to user {user_id}") else: logger.warning(f"Push notification failed: {response.status_code}") failed_subscriptions.append(subscription.id) except WebPushException as e: logger.error(f"WebPush error for user {user_id}: {str(e)}") if e.response and e.response.status_code in [410, 404]: # Subscription is no longer valid failed_subscriptions.append(subscription.id) except Exception as e: logger.error(f"Unexpected error sending push to user {user_id}: {str(e)}") failed_subscriptions.append(subscription.id) # Clean up failed subscriptions if failed_subscriptions: PushSubscription.query.filter( PushSubscription.id.in_(failed_subscriptions) ).update({'is_active': False}) db.session.commit() return successful_sends > 0 except Exception as e: logger.error(f"Failed to send push notification to user {user_id}: {str(e)}") return False def send_notification_to_multiple_users(self, user_ids: List[int], title: str, body: str, data: Optional[Dict] = None) -> Dict[str, int]: """Send push notification to multiple users""" results = {'success': 0, 'failed': 0} for user_id in user_ids: if self.send_notification_to_user(user_id, title, body, data): results['success'] += 1 else: results['failed'] += 1 return results def send_job_match_notification(self, worker_id: int, job_title: str, job_id: int) -> bool: """Send job match notification""" title = "New Job Match! 🎯" body = f"A new job matches your profile: {job_title}" data = { 'type': 'job_match', 'job_id': job_id, 'url': f'/jobs/{job_id}' } return self.send_notification_to_user(worker_id, title, body, data) def send_message_notification(self, user_id: int, sender_name: str, message_preview: str) -> bool: """Send new message notification""" title = f"New message from {sender_name}" body = message_preview[:100] + "..." if len(message_preview) > 100 else message_preview data = { 'type': 'message', 'url': '/messages' } return self.send_notification_to_user(user_id, title, body, data) def send_rating_notification(self, user_id: int, rating_value: int, contract_id: int) -> bool: """Send rating received notification""" stars = "⭐" * rating_value title = f"You received a {rating_value}-star rating! {stars}" body = "Check out your new rating and feedback" data = { 'type': 'rating', 'contract_id': contract_id, 'rating': rating_value, 'url': f'/contracts/{contract_id}' } return self.send_notification_to_user(user_id, title, body, data) def send_application_notification(self, contractor_id: int, worker_name: str, job_title: str) -> bool: """Send job application notification""" title = "New Job Application! 📋" body = f"{worker_name} applied for '{job_title}'" data = { 'type': 'application', 'url': '/applications' } return self.send_notification_to_user(contractor_id, title, body, data) def send_payment_notification(self, user_id: int, amount: float, contract_title: str) -> bool: """Send payment received notification""" title = "Payment Received! 💰" body = f"You received ${amount:.2f} for '{contract_title}'" data = { 'type': 'payment', 'amount': amount, 'url': '/dashboard' } return self.send_notification_to_user(user_id, title, body, data) def send_contract_update_notification(self, user_id: int, contract_title: str, status: str, contract_id: int) -> bool: """Send contract status update notification""" title = "Contract Update" body = f"'{contract_title}' is now {status}" data = { 'type': 'contract_update', 'contract_id': contract_id, 'status': status, 'url': f'/contracts/{contract_id}' } return self.send_notification_to_user(user_id, title, body, data) def send_reminder_notification(self, user_id: int, reminder_type: str, details: str) -> bool: """Send reminder notifications""" reminders = { 'profile_incomplete': { 'title': "Complete Your Profile", 'body': "Complete your profile to get more job matches" }, 'pending_rating': { 'title': "Rate Your Recent Work", 'body': f"Please rate your recent project: {details}" }, 'contract_deadline': { 'title': "Contract Deadline Approaching", 'body': f"Contract deadline in 24 hours: {details}" } } reminder = reminders.get(reminder_type) if not reminder: return False data = { 'type': 'reminder', 'reminder_type': reminder_type, 'url': '/dashboard' } return self.send_notification_to_user(user_id, reminder['title'], reminder['body'], data) def get_user_subscriptions(self, user_id: int) -> List[Dict]: """Get all active subscriptions for a user""" subscriptions = PushSubscription.query.filter_by( user_id=user_id, is_active=True ).all() return [sub.to_dict() for sub in subscriptions] def get_vapid_public_key(self) -> str: """Get VAPID public key for client-side subscription""" return self.vapid_public_key def cleanup_invalid_subscriptions(self) -> int: """Clean up subscriptions that are no longer valid""" try: # Deactivate old subscriptions (older than 6 months) six_months_ago = datetime.now(timezone.utc) - timedelta(days=180) old_subscriptions = PushSubscription.query.filter( PushSubscription.updated_at < six_months_ago, PushSubscription.is_active == True ).all() count = 0 for subscription in old_subscriptions: subscription.is_active = False count += 1 db.session.commit() logger.info(f"Cleaned up {count} old push subscriptions") return count except Exception as e: logger.error(f"Failed to cleanup push subscriptions: {str(e)}") return 0 # Integration with existing notification system class NotificationIntegration: """Integration layer between push notifications and existing notification system""" def __init__(self): self.push_service = PushNotificationService() def send_notification_with_push(self, user_id: int, notification_type: str, title: str, message: str, data: Optional[Dict] = None): """Send both in-app notification and push notification""" try: # Create in-app notification notification = Notification( user_id=user_id, type=notification_type, title=title, message=message, data=data or {} ) db.session.add(notification) db.session.commit() # Send push notification self.push_service.send_notification_to_user(user_id, title, message, data) except Exception as e: logger.error(f"Failed to send integrated notification: {str(e)}") db.session.rollback() # Utility functions for generating VAPID keys def generate_vapid_keys(): """Generate new VAPID keys for push notifications""" from pywebpush import vapid vapid_key = vapid.Vapid() vapid_key.generate_keys() return { 'private_key': vapid_key.private_key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.PKCS8, encryption_algorithm=serialization.NoEncryption() ).decode('utf-8'), 'public_key': vapid_key.public_key.public_bytes( encoding=serialization.Encoding.X962, format=serialization.PublicFormat.UncompressedPoint ) } # Export the service instance push_notification_service = PushNotificationService() notification_integration = NotificationIntegration()