import uuid import re from typing import List, Optional from sqlalchemy.orm import Session from app.models import User, Notification, Task, Comment, Mention class NotificationService: """Service for creating and managing notifications.""" MAX_MENTIONS_PER_COMMENT = 10 @staticmethod def create_notification( db: Session, user_id: str, notification_type: str, reference_type: str, reference_id: str, title: str, message: Optional[str] = None, ) -> Notification: """Create a notification for a user.""" notification = Notification( id=str(uuid.uuid4()), user_id=user_id, type=notification_type, reference_type=reference_type, reference_id=reference_id, title=title, message=message, ) db.add(notification) return notification @staticmethod def notify_task_assignment( db: Session, task: Task, assigned_by: User, ) -> Optional[Notification]: """Notify user when they are assigned to a task.""" if not task.assignee_id or task.assignee_id == assigned_by.id: return None return NotificationService.create_notification( db=db, user_id=task.assignee_id, notification_type="assignment", reference_type="task", reference_id=task.id, title=f"You've been assigned to: {task.title}", message=f"Assigned by {assigned_by.name}", ) @staticmethod def notify_blocker( db: Session, task: Task, reported_by: User, reason: str, ) -> List[Notification]: """Notify project owner when a task is blocked.""" notifications = [] # Notify project owner project = task.project if project and project.owner_id and project.owner_id != reported_by.id: notification = NotificationService.create_notification( db=db, user_id=project.owner_id, notification_type="blocker", reference_type="task", reference_id=task.id, title=f"Task blocked: {task.title}", message=f"Reported by {reported_by.name}: {reason[:100]}...", ) notifications.append(notification) return notifications @staticmethod def notify_blocker_resolved( db: Session, task: Task, resolved_by: User, reporter_id: str, ) -> Optional[Notification]: """Notify the original reporter when a blocker is resolved.""" if reporter_id == resolved_by.id: return None return NotificationService.create_notification( db=db, user_id=reporter_id, notification_type="blocker_resolved", reference_type="task", reference_id=task.id, title=f"Blocker resolved: {task.title}", message=f"Resolved by {resolved_by.name}", ) @staticmethod def count_mentions(content: str) -> int: """Count the number of @mentions in content.""" pattern = r'@([a-zA-Z0-9._-]+(?:@[a-zA-Z0-9.-]+)?)' matches = re.findall(pattern, content) return len(matches) @staticmethod def parse_mentions(content: str) -> List[str]: """Extract @mentions from comment content. Returns list of email usernames.""" # Match @username patterns (alphanumeric and common email chars before @domain) pattern = r'@([a-zA-Z0-9._-]+(?:@[a-zA-Z0-9.-]+)?)' matches = re.findall(pattern, content) return matches[:NotificationService.MAX_MENTIONS_PER_COMMENT] @staticmethod def process_mentions( db: Session, comment: Comment, task: Task, author: User, ) -> List[Notification]: """Process mentions in a comment and create notifications.""" notifications = [] mentioned_usernames = NotificationService.parse_mentions(comment.content) if not mentioned_usernames: return notifications # Find users by email or name for username in mentioned_usernames: # Try to find user by email first user = db.query(User).filter( (User.email.ilike(f"{username}%")) | (User.name.ilike(f"%{username}%")) ).first() if user and user.id != author.id: # Create mention record mention = Mention( id=str(uuid.uuid4()), comment_id=comment.id, mentioned_user_id=user.id, ) db.add(mention) # Create notification notification = NotificationService.create_notification( db=db, user_id=user.id, notification_type="mention", reference_type="comment", reference_id=comment.id, title=f"{author.name} mentioned you in: {task.title}", message=comment.content[:100] + ("..." if len(comment.content) > 100 else ""), ) notifications.append(notification) return notifications @staticmethod def notify_comment_reply( db: Session, comment: Comment, task: Task, author: User, parent_author_id: str, ) -> Optional[Notification]: """Notify original commenter when someone replies.""" if parent_author_id == author.id: return None return NotificationService.create_notification( db=db, user_id=parent_author_id, notification_type="comment", reference_type="comment", reference_id=comment.id, title=f"{author.name} replied to your comment on: {task.title}", message=comment.content[:100] + ("..." if len(comment.content) > 100 else ""), )