"""Notification service for security alerts and task reminders."""
import asyncio
import json
import logging
from datetime import date, datetime, timezone
from sqlalchemy import func, select, text
from fabledassistant.models import async_session
from fabledassistant.models.app_log import AppLog
from fabledassistant.models.note import Note
from fabledassistant.models.setting import Setting
from fabledassistant.models.user import User
from fabledassistant.services.email import _email_html, is_smtp_configured, send_email
from fabledassistant.services.logging import log_audit
logger = logging.getLogger(__name__)
_notification_task: asyncio.Task | None = None
SECURITY_EVENT_LABELS = {
"login": "New Login",
"login_failed": "Failed Login Attempt",
"logout": "Logout",
"password_change": "Password Changed",
}
async def _get_user_notification_pref(user_id: int, key: str) -> bool:
"""Check if a user has a notification preference enabled (default True)."""
async with async_session() as session:
result = await session.execute(
select(Setting).where(Setting.user_id == user_id, Setting.key == key)
)
setting = result.scalar_one_or_none()
# Default to enabled
return setting.value != "false" if setting else True
async def _get_user_email(user_id: int) -> str | None:
"""Get a user's email address."""
async with async_session() as session:
user = await session.get(User, user_id)
return user.email if user else None
async def notify_security_event(
user_id: int, event_type: str, details: dict | None = None
) -> None:
"""Send a security alert email to a user if they have alerts enabled and an email set."""
try:
if not await is_smtp_configured():
return
if not await _get_user_notification_pref(user_id, "notify_security_alerts"):
return
email = await _get_user_email(user_id)
if not email:
return
label = SECURITY_EVENT_LABELS.get(event_type, event_type)
ip = details.get("ip_address", "Unknown") if details else "Unknown"
timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
detail_rows = ""
if details:
for k, v in details.items():
if k == "ip_address":
continue
detail_rows += (
f''
f'| {k} | '
f'{v} | '
f'
'
)
body = f"""
{label}
| Time |
{timestamp} |
| IP Address |
{ip} |
{detail_rows}
If this wasn't you, change your password immediately.
"""
await send_email(email, f"Fabled Scribe - {label}", _email_html("Security Alert", body))
except Exception:
logger.exception("Failed to send security notification for user %d", user_id)
async def send_password_reset_email(email: str, reset_url: str) -> None:
"""Send a password reset email with a link to reset the user's password."""
body = f"""
We received a request to reset your password. Click the button below to choose a new password.
This link expires in 1 hour.
If you didn't request this, you can safely ignore this email.
If the button doesn't work, copy and paste this link:
{reset_url}
"""
await send_email(email, "Fabled Scribe - Password Reset", _email_html("Password Reset", body))
async def send_password_reset_success_email(email: str) -> None:
"""Send a notification that the password was successfully reset."""
timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
body = f"""
Your password was successfully changed at {timestamp}.
If you didn't make this change, please contact your administrator immediately.
"""
await send_email(email, "Fabled Scribe - Password Changed", _email_html("Password Changed", body))
async def send_invitation_email(email: str, invite_url: str, invited_by_username: str) -> None:
"""Send a branded invitation email with a registration link."""
body = f"""
{invited_by_username} has invited you to join Fabled Scribe.
Click the button below to create your account.
This invitation expires in 7 days.
If you weren't expecting this, you can safely ignore this email.
If the button doesn't work, copy and paste this link:
{invite_url}
"""
await send_email(email, "Fabled Scribe - You're Invited!", _email_html("You're Invited!", body))
async def check_due_tasks() -> None:
"""Check for tasks due today and send reminder emails."""
if not await is_smtp_configured():
return
today = date.today()
today_str = today.isoformat()
async with async_session() as session:
# Find tasks due today or overdue, not done
result = await session.execute(
select(Note)
.where(
Note.status.isnot(None),
Note.status != "done",
Note.due_date <= today,
)
)
tasks = result.scalars().all()
if not tasks:
return
# Group by user
tasks_by_user: dict[int, list[Note]] = {}
for task in tasks:
if task.user_id:
tasks_by_user.setdefault(task.user_id, []).append(task)
for user_id, user_tasks in tasks_by_user.items():
try:
# Check notification pref
if not await _get_user_notification_pref(user_id, "notify_task_reminders"):
continue
email = await _get_user_email(user_id)
if not email:
continue
# Dedup: check if we already sent a reminder today
dedup_result = await session.execute(
select(func.count(AppLog.id)).where(
AppLog.action == "task_reminder",
AppLog.user_id == user_id,
AppLog.created_at >= today_str,
)
)
if (dedup_result.scalar() or 0) > 0:
continue
# Build task list HTML
task_rows = ""
for task in user_tasks:
overdue = task.due_date < today if task.due_date else False
date_color = "#ef4444" if overdue else "#374151"
date_label = f'{task.due_date.isoformat()}' if task.due_date else ""
overdue_badge = ' (overdue)' if overdue else ""
task_rows += (
f''
f'| {task.title} | '
f'{date_label}{overdue_badge} | '
f'{task.status} | '
f'
'
)
body = f"""
You have {len(user_tasks)} task(s) due:
| Task |
Due |
Status |
{task_rows}
"""
await send_email(email, f"Fabled Scribe - {len(user_tasks)} Task(s) Due", _email_html("Task Reminders", body))
await log_audit(
"task_reminder",
user_id=user_id,
details={"task_count": len(user_tasks), "task_ids": [t.id for t in user_tasks]},
)
logger.info("Sent task reminder to user %d (%d tasks)", user_id, len(user_tasks))
except Exception:
logger.exception("Failed to send task reminder for user %d", user_id)
async def _notification_loop() -> None:
while True:
await asyncio.sleep(3600) # hourly
try:
await check_due_tasks()
except Exception:
logger.exception("Error in notification loop")
def start_notification_loop() -> None:
global _notification_task
if _notification_task is None or _notification_task.done():
_notification_task = asyncio.create_task(_notification_loop())
# ---------------------------------------------------------------------------
# In-app notifications (Notification model — shares, group events)
# ---------------------------------------------------------------------------
async def create_in_app_notification(user_id: int, notif_type: str, payload: dict):
"""Create an in-app Notification record."""
from fabledassistant.models.notification import Notification
async with async_session() as session:
n = Notification(user_id=user_id, type=notif_type, payload=payload)
session.add(n)
await session.commit()
await session.refresh(n)
return n
async def _fire_push_notif(user_id: int, title: str, body: str, url: str) -> None:
try:
from fabledassistant.services.push import send_push_notification
await send_push_notification(user_id, title, body, url=url)
except Exception:
logger.exception("Push notification failed for user %d", user_id)
async def _fire_share_email(user_id: int, subject: str, body_text: str) -> None:
try:
if not await is_smtp_configured():
return
async with async_session() as session:
user = await session.get(User, user_id)
if user and user.email:
html = _email_html(subject, f"{body_text.replace(chr(10), '
')}
")
await send_email(user.email, subject, html)
except Exception:
logger.exception("Share email notification failed for user %d", user_id)
async def _group_member_ids(group_id: int) -> list[int]:
from fabledassistant.models.group import GroupMembership
async with async_session() as session:
rows = (await session.execute(
select(GroupMembership.user_id).where(GroupMembership.group_id == group_id)
)).scalars().all()
return list(rows)
async def notify_project_shared(
project_id: int,
permission: str,
invited_by_user_id: int,
target_user_id: int | None,
target_group_id: int | None,
) -> None:
from fabledassistant.models.project import Project
async with async_session() as session:
project = await session.get(Project, project_id)
inviter = await session.get(User, invited_by_user_id)
if not project or not inviter:
return
recipients: list[int] = []
if target_user_id:
recipients = [target_user_id]
elif target_group_id:
recipients = [uid for uid in await _group_member_ids(target_group_id) if uid != invited_by_user_id]
url = f"/projects/{project_id}"
msg = f"{inviter.username} shared \"{project.title}\" with you as {permission}"
for uid in recipients:
await create_in_app_notification(uid, "project_shared", {
"project_id": project_id,
"project_title": project.title,
"permission": permission,
"invited_by": inviter.username,
"url": url,
})
asyncio.create_task(_fire_push_notif(uid, "Project shared with you", msg, url))
asyncio.create_task(_fire_share_email(uid, f"[Fabled] {inviter.username} shared a project with you", msg))
async def notify_note_shared(
note_id: int,
permission: str,
invited_by_user_id: int,
target_user_id: int | None,
target_group_id: int | None,
) -> None:
from fabledassistant.models.note import Note
async with async_session() as session:
note = await session.get(Note, note_id)
inviter = await session.get(User, invited_by_user_id)
if not note or not inviter:
return
recipients: list[int] = []
if target_user_id:
recipients = [target_user_id]
elif target_group_id:
recipients = [uid for uid in await _group_member_ids(target_group_id) if uid != invited_by_user_id]
route = "tasks" if note.is_task else "notes"
url = f"/{route}/{note_id}"
msg = f"{inviter.username} shared \"{note.title}\" with you as {permission}"
for uid in recipients:
await create_in_app_notification(uid, "note_shared", {
"note_id": note_id,
"note_title": note.title,
"permission": permission,
"invited_by": inviter.username,
"url": url,
})
asyncio.create_task(_fire_push_notif(uid, "Note shared with you", msg, url))
asyncio.create_task(_fire_share_email(uid, f"[Fabled] {inviter.username} shared a note with you", msg))
async def notify_group_added(
group_id: int, role: str, invited_by_user_id: int, target_user_id: int
) -> None:
from fabledassistant.models.group import Group
async with async_session() as session:
group = await session.get(Group, group_id)
inviter = await session.get(User, invited_by_user_id)
if not group or not inviter:
return
url = f"/groups/{group_id}"
msg = f"{inviter.username} added you to group \"{group.name}\" as {role}"
await create_in_app_notification(target_user_id, "group_added", {
"group_id": group_id,
"group_name": group.name,
"role": role,
"invited_by": inviter.username,
"url": url,
})
asyncio.create_task(_fire_push_notif(target_user_id, "Added to a group", msg, url))
asyncio.create_task(_fire_share_email(target_user_id, "[Fabled] You've been added to a group", msg))
async def list_in_app_notifications(user_id: int, unread_only: bool = True) -> list[dict]:
from fabledassistant.models.notification import Notification
async with async_session() as session:
q = select(Notification).where(Notification.user_id == user_id)
if unread_only:
q = q.where(Notification.read_at.is_(None))
q = q.order_by(Notification.created_at.desc()).limit(50)
rows = (await session.execute(q)).scalars().all()
return [n.to_dict() for n in rows]
async def unread_notification_count(user_id: int) -> int:
from fabledassistant.models.notification import Notification
async with async_session() as session:
result = await session.execute(
select(func.count()).where(
Notification.user_id == user_id,
Notification.read_at.is_(None),
)
)
return result.scalar() or 0
async def mark_notification_read(user_id: int, notification_id: int) -> bool:
from fabledassistant.models.notification import Notification
from datetime import timezone as tz
async with async_session() as session:
n = (await session.execute(
select(Notification).where(
Notification.id == notification_id,
Notification.user_id == user_id,
)
)).scalar_one_or_none()
if not n:
return False
from datetime import datetime
n.read_at = datetime.now(tz.utc)
await session.commit()
return True
async def mark_all_notifications_read(user_id: int) -> int:
from fabledassistant.models.notification import Notification
from datetime import datetime, timezone as tz
from sqlalchemy import update as sa_update
async with async_session() as session:
result = await session.execute(
sa_update(Notification)
.where(Notification.user_id == user_id, Notification.read_at.is_(None))
.values(read_at=datetime.now(tz.utc))
.returning(Notification.id)
)
await session.commit()
return len(result.fetchall())