Messenger/app/notifications/service.py

import json
import re

from sqlalchemy.ext.asyncio import AsyncSession

from app.chats.repository import is_chat_muted_for_user, list_chat_members
from app.media.repository import list_attachments_by_message_ids
from app.messages.models import Message
from app.notifications.repository import (
    create_notification_log,
    delete_push_device_token,
    list_user_notifications,
    upsert_push_device_token,
)
from app.notifications.schemas import (
    NotificationRead,
    NotificationRequest,
    PushTokenDeleteRequest,
    PushTokenUpsertRequest,
)
from app.notifications.tasks import send_mention_notification_task, send_push_notification_task
from app.users.repository import list_users_by_ids

_MENTION_RE = re.compile(r"@([A-Za-z0-9_]{3,50})")


def _extract_mentions(text: str | None) -> set[str]:
    if not text:
        return set()
    return {match.group(1).lower() for match in _MENTION_RE.finditer(text)}


def _message_preview_label(message: Message, image_preview_url: str | None) -> str:
    text_preview = (message.text or "").strip()
    if text_preview:
        return text_preview[:120]
    message_type = message.type.value if hasattr(message.type, "value") else str(message.type)
    return {
        "image": "Photo" if image_preview_url else "Image",
        "video": "Video",
        "audio": "Audio",
        "voice": "Voice message",
        "circle_video": "Video note",
        "file": "File",
    }.get(message_type, "Message")


async def enqueue_notification(db: AsyncSession, payload: NotificationRequest) -> None:
    await create_notification_log(
        db,
        user_id=payload.user_id,
        event_type=payload.event_type,
        payload=json.dumps(payload.payload, ensure_ascii=True),
    )


async def dispatch_message_notifications(db: AsyncSession, message: Message) -> None:
    members = await list_chat_members(db, chat_id=message.chat_id)
    recipient_ids = [m.user_id for m in members if m.user_id != message.sender_id]
    if not recipient_ids:
        return

    users = await list_users_by_ids(db, recipient_ids)
    user_by_username = {user.username.lower(): user for user in users}
    mentioned_usernames = _extract_mentions(message.text)
    mentioned_user_ids = {user_by_username[name].id for name in mentioned_usernames if name in user_by_username}

    sender_users = await list_users_by_ids(db, [message.sender_id])
    sender_name = sender_users[0].username if sender_users else "Someone"
    attachments = await list_attachments_by_message_ids(db, message_ids=[message.id])
    first_attachment = attachments[0] if attachments else None
    preview_image_url = None
    if first_attachment and first_attachment.file_type.lower().startswith("image/"):
        preview_image_url = first_attachment.file_url
    preview_body = _message_preview_label(message, preview_image_url)
    message_type = message.type.value if hasattr(message.type, "value") else str(message.type)

    for recipient in users:
        base_payload = {
            "chat_id": message.chat_id,
            "message_id": message.id,
            "sender_id": message.sender_id,
            "message_type": message_type,
            "preview_image_url": preview_image_url or "",
            "sender_name": sender_name,
        }
        if recipient.id in mentioned_user_ids:
            payload = {
                **base_payload,
                "type": "mention",
                "text_preview": preview_body,
            }
            await create_notification_log(
                db,
                user_id=recipient.id,
                event_type="mention",
                payload=json.dumps(payload, ensure_ascii=True),
            )
            send_mention_notification_task.delay(
                recipient.id,
                sender_name,
                preview_body,
                payload,
            )
            continue

        if await is_chat_muted_for_user(db, chat_id=message.chat_id, user_id=recipient.id):
            continue

        payload = {
            **base_payload,
            "type": "message",
            "text_preview": preview_body,
        }
        await create_notification_log(
            db,
            user_id=recipient.id,
            event_type="message",
            payload=json.dumps(payload, ensure_ascii=True),
        )
        send_push_notification_task.delay(
            recipient.id,
            sender_name,
            preview_body,
            payload,
        )

    await db.commit()


async def get_notifications_for_user(db: AsyncSession, *, user_id: int, limit: int = 50) -> list[NotificationRead]:
    safe_limit = max(1, min(limit, 100))
    rows = await list_user_notifications(db, user_id=user_id, limit=safe_limit)
    return [NotificationRead.model_validate(item) for item in rows]


async def register_push_token(db: AsyncSession, *, user_id: int, payload: PushTokenUpsertRequest) -> None:
    await upsert_push_device_token(
        db,
        user_id=user_id,
        platform=payload.platform.strip().lower(),
        token=payload.token.strip(),
        device_id=payload.device_id.strip() if payload.device_id else None,
        app_version=payload.app_version.strip() if payload.app_version else None,
    )
    await db.commit()


async def unregister_push_token(db: AsyncSession, *, user_id: int, payload: PushTokenDeleteRequest) -> None:
    await delete_push_device_token(
        db,
        user_id=user_id,
        platform=payload.platform.strip().lower(),
        token=payload.token.strip(),
    )
    await db.commit()