89 lines
3.2 KiB
Python
89 lines
3.2 KiB
Python
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
|
|
)
|