Solo_bot/database/web_notifications.py

from datetime import UTC, datetime

from sqlalchemy import delete, func, select, update
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.ext.asyncio import AsyncSession

from database.models import User, WebNotification, WebPushSubscription
from logger import logger


async def upsert_push_subscription(
    session: AsyncSession,
    *,
    user_id: int,
    identity_id: str | None,
    endpoint: str,
    keys_json: dict,
) -> WebPushSubscription:
    """Upsert push subscription by endpoint (unique)."""
    stmt = (
        pg_insert(WebPushSubscription)
        .values(
            user_id=user_id,
            identity_id=identity_id,
            endpoint=endpoint,
            keys_json=keys_json,
            created_at=datetime.now(UTC),
        )
        .on_conflict_do_update(
            index_elements=["endpoint"],
            set_={
                "user_id": user_id,
                "identity_id": identity_id,
                "keys_json": keys_json,
                "created_at": datetime.now(UTC),
            },
        )
        .returning(WebPushSubscription)
    )
    result = await session.execute(stmt)
    return result.scalar_one()


async def get_push_subscriptions_by_user(
    session: AsyncSession,
    user_id: int,
) -> list[WebPushSubscription]:
    result = await session.execute(select(WebPushSubscription).where(WebPushSubscription.user_id == user_id))
    return list(result.scalars().all())


async def get_push_subscriptions_by_identity(
    session: AsyncSession,
    identity_id: str,
) -> list[WebPushSubscription]:
    result = await session.execute(select(WebPushSubscription).where(WebPushSubscription.identity_id == identity_id))
    return list(result.scalars().all())


async def delete_push_subscription_by_endpoint(
    session: AsyncSession,
    endpoint: str,
) -> None:
    await session.execute(delete(WebPushSubscription).where(WebPushSubscription.endpoint == endpoint))


async def get_notifications_for_identity(
    session: AsyncSession,
    identity_id: str,
    limit: int = 20,
    offset: int = 0,
) -> list[WebNotification]:
    result = await session.execute(
        select(WebNotification)
        .where(WebNotification.identity_id == identity_id)
        .order_by(WebNotification.created_at.desc())
        .limit(limit)
        .offset(offset)
    )
    return list(result.scalars().all())


async def count_unread_for_identity(
    session: AsyncSession,
    identity_id: str,
) -> int:
    result = await session.execute(
        select(func.count())
        .select_from(WebNotification)
        .where(
            WebNotification.identity_id == identity_id,
            WebNotification.read == False,  # noqa: E712 SQLAlchemy expression
        )
    )
    return result.scalar() or 0


async def mark_all_read_for_identity(
    session: AsyncSession,
    identity_id: str,
) -> int:
    result = await session.execute(
        update(WebNotification)
        .where(
            WebNotification.identity_id == identity_id,
            WebNotification.read == False,  # noqa: E712 SQLAlchemy expression
        )
        .values(read=True)
    )
    return result.rowcount


async def resolve_identity_id_by_tg_id(
    session: AsyncSession,
    tg_id: int,
) -> str | None:
    """Resolve identity_id from user's tg_id."""
    result = await session.execute(select(User.identity_id).where(User.tg_id == tg_id))
    return result.scalar_one_or_none()


async def create_notification(
    session: AsyncSession,
    *,
    user_id: int,
    identity_id: str | None,
    type: str = "system",
    title: str,
    message: str = "",
    data: dict | None = None,
) -> WebNotification:
    notif = WebNotification(
        user_id=user_id,
        identity_id=identity_id,
        type=type,
        title=title,
        message=message,
        data=data,
    )
    session.add(notif)
    await session.flush()
    return notif


def _render_template(template: str, **kwargs: object) -> str:
    """Safe format — unknown placeholders stay as-is."""
    try:
        return template.format_map(
            {k: str(v) for k, v in kwargs.items() if v is not None}
            | type("_Defaults", (), {"__missing__": lambda self, k: f"{{{k}}}"})()
        )
    except Exception:
        return template


def _get_web_config_str(key: str, default: str) -> str:
    try:
        from core.settings.web_config import WEB_CONFIG

        val = WEB_CONFIG.get(key)
        return str(val).strip() if val else default
    except Exception:
        return default


async def notify_web(
    session: AsyncSession,
    *,
    tg_id: int,
    type: str = "system",
    title: str | None = None,
    message: str | None = None,
    data: dict | None = None,
    template_vars: dict | None = None,
) -> WebNotification | None:
    """Создаёт web-уведомление по tg_id.

    title/message — если None, берутся из WEB_CONFIG шаблонов по type.
    template_vars — подстановки в шаблон ({email}, {amount}, {name}, {duration}).
    """
    try:
        identity_id = await resolve_identity_id_by_tg_id(session, tg_id)
        if not identity_id:
            return None

        vars_ = template_vars or {}

        type_key_map = {
            "payment": ("WEB_NOTIFY_PAYMENT_TITLE", "WEB_NOTIFY_PAYMENT_MESSAGE"),
            "key_created": ("WEB_NOTIFY_KEY_CREATED_TITLE", "WEB_NOTIFY_KEY_CREATED_MESSAGE"),
            "key_expiry": ("WEB_NOTIFY_KEY_EXPIRY_TITLE", "WEB_NOTIFY_KEY_EXPIRY_MESSAGE"),
            "gift_received": ("WEB_NOTIFY_GIFT_TITLE", "WEB_NOTIFY_GIFT_MESSAGE"),
        }
        title_key, msg_key = type_key_map.get(type, (None, None))

        resolved_title = title
        if resolved_title is None and title_key:
            resolved_title = _render_template(_get_web_config_str(title_key, ""), **vars_)
        resolved_title = resolved_title or type

        resolved_message = message
        if resolved_message is None and msg_key:
            resolved_message = _render_template(_get_web_config_str(msg_key, ""), **vars_)
        resolved_message = resolved_message or ""

        notif = await create_notification(
            session,
            user_id=tg_id,
            identity_id=identity_id,
            type=type,
            title=resolved_title,
            message=resolved_message,
            data=data,
        )

        try:
            from services.web_push import push_enabled, send_push_to_many

            if push_enabled():
                subs = await get_push_subscriptions_by_identity(session, identity_id)
                if subs:
                    sub_infos = [{"endpoint": s.endpoint, "keys": s.keys_json} for s in subs]
                    sent = await send_push_to_many(
                        sub_infos,
                        title=resolved_title,
                        body=resolved_message,
                        url="/dashboard/notifications",
                    )
                    logger.debug("[notify_web] push sent to {}/{} subscriptions", sent, len(sub_infos))
        except Exception as push_err:
            logger.warning("[notify_web] push delivery failed: {}", push_err)

        return notif
    except Exception:
        return None