HR-ATS-Portal/backend/notifications/views.py

from fastapi import HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
import httpx

import uuid

from notifications.models import EmailConfirmationTokens,Notifications
from notifications.plugins import (
    CONFIRM_TOKEN_RESEND_SECONDS,
    CONFIRM_TOKEN_TTL_SECONDS,
    build_confirmation_link,
    compose_token,
    confirmation_expiry,
    generate_token_secret,
    hash_token,
    now_utc,
    render_confirmation_email,
    send_confirmation_mail,
    split_token,
    verify_token,
)
from notifications.serializers import (
    serialize_confirmation_request,
    serialize_confirmation_result,
    serialize_notification,
)
from users.models import Users


class Confirmation:
    def __init__(self,session:AsyncSession):
        self.session=session

    async def send_confirmation(self,user):
        """Issue a fresh token for an already committed Users row and mail the link."""
        await EmailConfirmationTokens.invalidate_tokens_for_user(self.session,str(user.id))

        secret=generate_token_secret()
        expires_at=confirmation_expiry()
        row=await EmailConfirmationTokens.insert_token(self.session,{
            "user_id":user.id,
            "email":user.email,
            "token_hash":hash_token(secret),
            "expires_at":expires_at,
        })

        link=build_confirmation_link(compose_token(row.id,secret))
        subject,html=render_confirmation_email(link,CONFIRM_TOKEN_TTL_SECONDS)
        try:
            await send_confirmation_mail(user.email,subject,html)
        except (httpx.HTTPError,RuntimeError) as e:
            await EmailConfirmationTokens.mark_used(self.session,str(row.id))
            raise HTTPException(status_code=502,detail="Failed to send confirmation email") from e

        return serialize_confirmation_request(user.email,expires_at)

    async def confirm(self,token):
        record_id,secret=split_token(token)
        if not record_id:
            raise HTTPException(status_code=400,detail="Invalid confirmation link")

        row=await EmailConfirmationTokens.get_token_by_id(self.session,record_id)
        if not row or not verify_token(secret,row.token_hash):
            raise HTTPException(status_code=400,detail="Invalid confirmation link")

        user=await Users.get_user_by_id(self.session,str(row.user_id))
        if not user or user.is_deleted:
            raise HTTPException(status_code=404,detail="User not found")

        # Mail clients, link scanners and the back button all replay this link.
        if row.is_used:
            if row.confirmed_at and user.is_active:
                return serialize_confirmation_result(user,already_confirmed=True)
            raise HTTPException(status_code=400,detail="This confirmation link is no longer valid. Request a new one.")

        if row.expires_at<=now_utc():
            raise HTTPException(status_code=400,detail="Confirmation link has expired. Request a new one.")

        if user.is_active:
            await EmailConfirmationTokens.mark_confirmed(self.session,str(row.id))
            return serialize_confirmation_result(user,already_confirmed=True)

        updated=await Users.update_user(self.session,str(user.id),{"is_active":True})
        await EmailConfirmationTokens.mark_confirmed(self.session,str(row.id))
        return serialize_confirmation_result(updated)

    async def resend(self,email):
        user=await Users.get_user_by_email(self.session,email)
        if not user or user.is_deleted:
            raise HTTPException(status_code=404,detail="No account found for this email")
        if user.is_active:
            raise HTTPException(status_code=400,detail="This account is already confirmed")

        active=await EmailConfirmationTokens.get_active_token_by_user(self.session,str(user.id))
        if active:
            age=(now_utc()-active.created_at).total_seconds()
            if age<CONFIRM_TOKEN_RESEND_SECONDS and active.expires_at>now_utc():
                raise HTTPException(status_code=429,detail="Please wait before requesting another confirmation email")

        return await self.send_confirmation(user)


def _as_uuid(value):
    if value in (None,""):
        return None
    try:
        return uuid.UUID(str(value))
    except (TypeError,ValueError):
        return None


def _user_id(current_user):
    if not current_user or not current_user.get("id"):
        raise HTTPException(status_code=401,detail="Not authenticated")
    uid=_as_uuid(current_user["id"])
    if uid is None:
        raise HTTPException(status_code=401,detail="Invalid user id")
    return uid


class Notification:
    def __init__(self,session:AsyncSession):
        self.session=session

    async def get_notifications(self,current_user,unread_only=False,top=None,skip=0):
        rows,total,unread=await Notifications.fetch_notifications(
            self.session,
            user_id=_user_id(current_user),
            unread_only=bool(unread_only),
            top=top,
            skip=skip or 0,
        )
        return [serialize_notification(r) for r in rows],total,unread

    async def mark_read(self,record_id,current_user):
        row=await Notifications.mark_read(
            self.session,record_id,user_id=_user_id(current_user)
        )
        if not row:
            raise HTTPException(status_code=404,detail="Notification not found")
        return serialize_notification(row)

    async def mark_all_read(self,current_user):
        count=await Notifications.mark_all_read(self.session,_user_id(current_user))
        return {"updated": count}

    async def delete_notification(self,record_id,current_user):
        row=await Notifications.soft_delete_notification(
            self.session,record_id,user_id=_user_id(current_user)
        )
        if not row:
            raise HTTPException(status_code=404,detail="Notification not found")
        return {"id": str(row.id),"deleted": True}