arrr-erp/app/modules/alerts/service.py

from __future__ import annotations

from datetime import datetime, timezone
from typing import Iterable

from sqlalchemy import Select, func, select, update
from sqlalchemy.orm import Session

from app.modules.alerts.models import UserAlert
from app.modules.core.iam.models import User
from app.modules.email_integration.event_service import send_alert_created_email

ALERT_PRIORITIES = ("low", "normal", "high", "critical")
ALERT_TYPES = (
    "general",
    "task_assigned",
    "task_due",
    "task_overdue",
    "task_review",
    "document_uploaded",
    "clarification",
    "attendance",
    "leave",
    "payroll",
    "consultant",
    "client",
)


def normalize_priority(priority: str | None) -> str:
    value = (priority or "normal").strip().lower()
    return value if value in ALERT_PRIORITIES else "normal"


def normalize_alert_type(alert_type: str | None) -> str:
    value = (alert_type or "general").strip().lower()
    return value or "general"


def create_alert(
    db: Session,
    *,
    user_id: int,
    title: str,
    message: str | None = None,
    tenant_id: int | None = None,
    branch_id: int | None = None,
    role_context: str | None = None,
    alert_type: str = "general",
    priority: str = "normal",
    target_url: str | None = None,
    created_by_user_id: int | None = None,
    commit: bool = True,
) -> UserAlert:
    alert = UserAlert(
        tenant_id=tenant_id,
        branch_id=branch_id,
        user_id=user_id,
        role_context=(role_context or None),
        alert_type=normalize_alert_type(alert_type),
        priority=normalize_priority(priority),
        title=(title or "Alert").strip()[:255],
        message=(message or None),
        target_url=(target_url or None),
        created_by_user_id=created_by_user_id,
    )
    db.add(alert)
    db.flush()
    try:
        send_alert_created_email(db, alert)
    except Exception:
        # Email notification must never block in-app alert creation.
        pass
    if commit:
        db.commit()
        db.refresh(alert)
    return alert


def create_bulk_alerts(
    db: Session,
    *,
    user_ids: Iterable[int],
    title: str,
    message: str | None = None,
    tenant_id: int | None = None,
    branch_id: int | None = None,
    role_context: str | None = None,
    alert_type: str = "general",
    priority: str = "normal",
    target_url: str | None = None,
    created_by_user_id: int | None = None,
) -> list[UserAlert]:
    rows: list[UserAlert] = []
    for user_id in sorted({int(uid) for uid in user_ids if uid}):
        rows.append(
            create_alert(
                db,
                user_id=user_id,
                title=title,
                message=message,
                tenant_id=tenant_id,
                branch_id=branch_id,
                role_context=role_context,
                alert_type=alert_type,
                priority=priority,
                target_url=target_url,
                created_by_user_id=created_by_user_id,
                commit=False,
            )
        )
    db.commit()
    for row in rows:
        db.refresh(row)
    return rows


def _user_alert_query(current_user: User) -> Select:
    return select(UserAlert).where(UserAlert.user_id == current_user.id)


def list_my_alerts(
    db: Session,
    current_user: User,
    *,
    status: str = "all",
    priority: str = "all",
    limit: int = 100,
) -> list[UserAlert]:
    q = _user_alert_query(current_user)
    if status == "unread":
        q = q.where(UserAlert.is_read.is_(False))
    elif status == "read":
        q = q.where(UserAlert.is_read.is_(True))
    if priority in ALERT_PRIORITIES:
        q = q.where(UserAlert.priority == priority)
    q = q.order_by(UserAlert.is_read.asc(), UserAlert.created_at_utc.desc()).limit(max(1, min(limit, 500)))
    return list(db.execute(q).scalars().all())


def count_unread_alerts(db: Session, current_user: User | None) -> int:
    if not current_user:
        return 0
    value = db.execute(
        select(func.count(UserAlert.id)).where(UserAlert.user_id == current_user.id, UserAlert.is_read.is_(False))
    ).scalar_one()
    return int(value or 0)


def get_my_alert_or_404(db: Session, current_user: User, alert_id: int) -> UserAlert | None:
    return db.execute(
        select(UserAlert).where(UserAlert.id == alert_id, UserAlert.user_id == current_user.id)
    ).scalar_one_or_none()


def mark_alert_read(db: Session, current_user: User, alert_id: int) -> bool:
    alert = get_my_alert_or_404(db, current_user, alert_id)
    if not alert:
        return False
    if not alert.is_read:
        alert.is_read = True
        alert.read_at_utc = datetime.now(timezone.utc)
        db.commit()
    return True


def mark_all_alerts_read(db: Session, current_user: User) -> int:
    result = db.execute(
        update(UserAlert)
        .where(UserAlert.user_id == current_user.id, UserAlert.is_read.is_(False))
        .values(is_read=True, read_at_utc=datetime.now(timezone.utc))
    )
    db.commit()
    return int(result.rowcount or 0)