from typing import List, Optional from sqlmodel import Session, select from models import Notification, NotificationType, User, UserVerticalPreference, Show, PreferenceTier class NotificationService: def __init__(self, session: Session): self.session = session def create_notification( self, user_id: int, type: NotificationType, title: str, message: str, link: Optional[str] = None ) -> Notification: notification = Notification( user_id=user_id, type=type, title=title, message=message, link=link ) self.session.add(notification) self.session.commit() self.session.refresh(notification) return notification def get_user_notifications( self, user_id: int, limit: int = 50, offset: int = 0 ) -> List[Notification]: query = select(Notification).where(Notification.user_id == user_id).order_by(Notification.created_at.desc()).offset(offset).limit(limit) return self.session.exec(query).all() def mark_as_read(self, notification_id: int, user_id: int) -> bool: notification = self.session.get(Notification, notification_id) if notification and notification.user_id == user_id: notification.is_read = True self.session.add(notification) self.session.commit() return True return False def mark_all_as_read(self, user_id: int): statement = select(Notification).where(Notification.user_id == user_id).where(Notification.is_read == False) unread = self.session.exec(statement).all() for note in unread: note.is_read = True self.session.add(note) self.session.commit() def check_show_alert(self, show: Show): """ Check if any users want to be notified about this new show. This roughly matches users who: 1. Follow the vertical (UserVerticalPreference) 2. Have notify_on_show = True 3. Valid Tier (Usually Headliner/MainStage) """ # Find users who subscribe to this band # Filtering logic: # - Matches vertical_id # - notify_on_show is True # - Tier is HEADLINER (for high priority) or specific preference # For now, let's alert all who have notify_on_show=True for this vertical subscriptions = self.session.exec( select(UserVerticalPreference) .where(UserVerticalPreference.vertical_id == show.vertical_id) .where(UserVerticalPreference.notify_on_show == True) ).all() for sub in subscriptions: # We can customize message based on tier if needed title = f"New Show Added: {show.vertical.name}" date_str = show.date.strftime("%b %d, %Y") message = f"{show.vertical.name} at {show.venue.name} on {date_str}" link = f"/{show.vertical.slug}/shows/{show.slug}" self.create_notification( user_id=sub.user_id, type=NotificationType.SHOW_ALERT, title=title, message=message, link=link )