"""Notification service for security alerts and task reminders."""
import asyncio
import json
import logging
from datetime import date, datetime, time, timezone
from sqlalchemy import func, select, text
from scribe.models import async_session
from scribe.models.app_log import AppLog
from scribe.models.note import Note
from scribe.models.setting import Setting
from scribe.models.user import User
from scribe.services.email import _email_html, is_smtp_configured, send_email
from scribe.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))
def utc_day_start(day: date) -> datetime:
"""Midnight UTC on `day`, as an AWARE datetime — the reminder dedup window.
Deliberately a `datetime` and not the bare `date` (#1727). Comparing a
`timestamptz` column against a `date` does work, via an implicit cast — but
Postgres resolves that cast in the SESSION's TimeZone, so the window would
move with a server setting nobody remembers is load-bearing. An aware UTC
datetime says what it means and compares the same way everywhere.
"""
return datetime.combine(day, time.min, tzinfo=timezone.utc)
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()
window_start = utc_day_start(today)
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.
# `window_start` is an aware datetime, NOT `today.isoformat()`.
# asyncpg binds a str as VARCHAR and Postgres has no
# `timestamptz >= text` operator, so this raised on every run
# that got this far — swallowed by the per-user `except` below,
# which is why the only symptom was reminders silently never
# sending plus an hourly traceback in the DB log (#1727).
dedup_result = await session.execute(
select(func.count(AppLog.id)).where(
AppLog.action == "task_reminder",
AppLog.user_id == user_id,
AppLog.created_at >= window_start,
)
)
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)
# Read notifications are kept this long before the hourly sweep deletes them;
# unread are kept regardless. Without a sweep the table grows without bound.
_NOTIFICATION_RETENTION_DAYS = 30
async def purge_old_read_notifications(retention_days: int = _NOTIFICATION_RETENTION_DAYS) -> int:
"""Delete already-read in-app notifications older than retention_days."""
from datetime import timedelta
from sqlalchemy import delete
from scribe.models.notification import Notification
cutoff = datetime.now(timezone.utc) - timedelta(days=retention_days)
async with async_session() as session:
result = await session.execute(
delete(Notification).where(
Notification.read_at.isnot(None),
Notification.read_at < cutoff,
)
)
await session.commit()
return result.rowcount or 0
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")
try:
removed = await purge_old_read_notifications()
if removed:
logger.info("Notification retention: deleted %d read notification(s)", removed)
except Exception:
logger.exception("Error in notification retention cleanup")
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 scribe.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_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 scribe.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 scribe.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_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 scribe.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_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 scribe.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_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 scribe.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 scribe.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 scribe.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 scribe.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())