import asyncio
import hashlib
import hmac
import json
import os
import re
import unicodedata
import uuid
from datetime import UTC, datetime, timedelta
from pathlib import PurePath
from typing import Any

from sqlalchemy import case, func, select, update
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession

from app.db import Document, UserIdentity
from app.integrations import S3Client, SafetyClient
from app.notification_models import (
    ClientDocument,
    ClientUploadDraft,
    GuestNotification,
    Notification,
    NotificationButton,
    NotificationCtaAction,
    NotificationDocument,
    NotificationSource,
    NotificationType,
    uuid7,
)
from app.notification_schemas import (
    NotificationCancelRequest,
    NotificationCreateRequest,
    UploadCompleteRequest,
    UploadInitRequest,
)
from app.realtime import RealtimeFanout
from app.services import AuditContext, DomainError, SettingsSnapshot, audit


def fingerprint(body: NotificationCreateRequest) -> str:
    canonical = json.dumps(
        body.model_dump(mode="json"),
        sort_keys=True,
        separators=(",", ":"),
        ensure_ascii=False,
    )
    return hashlib.sha256(canonical.encode()).hexdigest()


def source_token_hash(token: str) -> str:
    return hashlib.sha256(token.encode()).hexdigest()


async def synchronize_source_tokens(session: AsyncSession) -> None:
    token = os.getenv("NOTIFICATIONS_TOKEN_PRODUCER_TEST")
    if not token:
        return
    source = await session.scalar(
        select(NotificationSource).where(NotificationSource.code == "producer_test")
    )
    digest = source_token_hash(token)
    if source and not hmac.compare_digest(source.token_hash, digest):
        source.token_hash = digest
        source.token_rotated_at = datetime.now(UTC)
        await session.commit()


async def authenticate_source(
    session: AsyncSession, authorization: str | None
) -> NotificationSource:
    if not authorization or not authorization.startswith("Bearer "):
        raise DomainError("unauthorized", 401, "Authentication failed")
    token = authorization.removeprefix("Bearer ").strip()
    if not token:
        raise DomainError("unauthorized", 401, "Authentication failed")
    supplied = source_token_hash(token)
    sources = (
        (
            await session.execute(
                select(NotificationSource).where(NotificationSource.record_status == "A")
            )
        )
        .scalars()
        .all()
    )
    matched: NotificationSource | None = None
    for source in sources:
        if hmac.compare_digest(source.token_hash, supplied):
            matched = source
    if matched is None:
        raise DomainError("unauthorized", 401, "Authentication failed")
    return matched


async def _type(session: AsyncSession, code: str, contour: str | None = None) -> NotificationType:
    query = select(NotificationType).where(
        NotificationType.code == code, NotificationType.record_status == "A"
    )
    if contour:
        query = query.where(NotificationType.contour == contour)
    item = await session.scalar(query)
    if item is None:
        raise DomainError(
            "validation_error",
            400,
            "Notification type is unavailable",
            {"fields": ["notification_type"]},
        )
    return item


def _details_dict(body: NotificationCreateRequest) -> dict[str, Any] | None:
    return (
        body.details.model_dump(mode="json", exclude_none=True, exclude_defaults=True)
        if body.details
        else None
    )


def validate_create(
    body: NotificationCreateRequest,
    kind: NotificationType,
    action: NotificationCtaAction,
) -> None:
    values: dict[str, Any] = {
        "details": body.details,
        "payment_url": body.payment_url,
        "chat_message_text": body.chat_message_text,
    }
    required_fields = set(action.required_instance_fields)
    errors: list[str] = []
    for field in values:
        if field in required_fields and values[field] is None:
            errors.append(field)
        elif field not in required_fields and values[field] is not None:
            errors.append(field)
    details = _details_dict(body)
    if "details" in required_fields and not details:
        errors.append("details")
    if details is not None:
        for block in kind.required_detail_blocks:
            if not details.get(block):
                errors.append(f"details.{block}")
        documents = details.get("documents") or []
        if documents and not kind.documents_allowed:
            errors.append("details.documents")
        if details.get("send_documents") and "send_docs" not in {
            kind.button_primary_code,
            kind.button_secondary_code,
        }:
            errors.append("details.send_documents")
    if errors:
        raise DomainError(
            "validation_error",
            400,
            "Notification fields do not match catalog rules",
            {"fields": sorted(set(errors))},
        )


async def create_notification(
    session: AsyncSession,
    body: NotificationCreateRequest,
    source: NotificationSource,
    s3: S3Client,
    fanout: RealtimeFanout,
    context: AuditContext,
) -> tuple[dict[str, Any], int]:
    if body.source != source.code:
        raise DomainError("forbidden", 403, "Source does not match service token")
    digest = fingerprint(body)
    existing = await session.scalar(
        select(Notification).where(
            Notification.source == source.code,
            Notification.external_id == body.external_id,
        )
    )
    if existing:
        if hmac.compare_digest(existing.request_fingerprint, digest):
            return await notification_dto(session, existing), 200
        session.add(
            audit(
                "notification.create_conflict",
                context,
                None,
                "notification",
                existing.id,
                metadata={"source": source.code},
                outcome="failed",
            )
        )
        await session.commit()
        raise DomainError(
            "notification_conflict",
            409,
            "External id already belongs to another request",
            {"notification_id": str(existing.id)},
        )
    kind = await _type(session, body.notification_type, "P")
    action = await session.scalar(
        select(NotificationCtaAction).where(
            NotificationCtaAction.code == kind.cta_action,
            NotificationCtaAction.record_status == "A",
        )
    )
    if action is None:
        raise DomainError("validation_error", 400, "Notification CTA is unavailable")
    validate_create(body, kind, action)
    if (
        await session.scalar(
            select(UserIdentity.id).where(
                UserIdentity.id == body.user_id, UserIdentity.record_status == "A"
            )
        )
        is None
    ):
        raise DomainError("validation_error", 400, "User is unavailable", {"fields": ["user_id"]})
    details = _details_dict(body)
    document_inputs = list(details.pop("documents", [])) if details else []
    item = Notification(
        id=uuid7(),
        user_id=body.user_id,
        notification_type=body.notification_type,
        source=body.source,
        external_id=body.external_id,
        request_fingerprint=digest,
        notification_datetime=body.notification_datetime,
        header=body.header,
        text=body.text,
        priority_override=body.priority_override,
        date_expired=body.date_expired,
        price=body.price,
        old_price=body.old_price,
        payment_url=str(body.payment_url) if body.payment_url else None,
        details=details,
        chat_message_text=body.chat_message_text,
    )
    session.add(item)
    for order, document_input in enumerate(document_inputs):
        bucket = s3.settings.selectel_s3_bucket_documents
        try:
            metadata = await s3.head(bucket, document_input["object_key"])
        except Exception as exc:
            raise DomainError(
                "validation_error",
                400,
                "Company document object is unavailable",
                {"fields": [f"details.documents.{order}.object_key"]},
            ) from exc
        if (
            int(metadata["ContentLength"]) != document_input["size_bytes"]
            or metadata.get("ContentType") != document_input["mime_type"]
        ):
            raise DomainError(
                "validation_error",
                400,
                "Company document metadata differs",
                {"fields": [f"details.documents.{order}"]},
            )
        document = await session.scalar(
            select(Document).where(
                Document.storage_bucket == bucket,
                Document.object_key == document_input["object_key"],
            )
        )
        if document is None:
            document = Document(
                id=uuid7(),
                user_id=body.user_id,
                name=document_input["title"],
                mime_type=document_input["mime_type"],
                size_bytes=document_input["size_bytes"],
                checksum_sha256=document_input["checksum_sha256"],
                storage_bucket=bucket,
                object_key=document_input["object_key"],
                sent_at=datetime.now(UTC),
            )
            session.add(document)
        elif document.user_id != body.user_id:
            raise DomainError("validation_error", 400, "Company document belongs to another user")
        session.add(
            NotificationDocument(
                id=uuid7(),
                notification_id=item.id,
                document_id=document.id,
                sort_order=order,
            )
        )
    session.add(
        audit(
            "notification.created",
            context,
            None,
            "notification",
            item.id,
            metadata={"source": source.code, "notification_type": item.notification_type},
        )
    )
    try:
        await session.commit()
    except IntegrityError:
        await session.rollback()
        raced = await session.scalar(
            select(Notification).where(
                Notification.source == source.code,
                Notification.external_id == body.external_id,
            )
        )
        if raced and hmac.compare_digest(raced.request_fingerprint, digest):
            return await notification_dto(session, raced), 200
        raise DomainError("notification_conflict", 409, "External id is already used") from None
    result = await notification_dto(session, item)
    unread = await unread_count(session, item.user_id, 15)
    await fanout.publish_user(
        item.user_id,
        {
            "type": "notification.created",
            "occurred_at": datetime.now(UTC).isoformat(),
            "notification": result,
            "unread_count": unread,
        },
    )
    return result, 201


async def cancel_notification(
    session: AsyncSession,
    body: NotificationCancelRequest,
    source: NotificationSource,
    fanout: RealtimeFanout,
    context: AuditContext,
) -> dict[str, Any]:
    if body.source != source.code:
        raise DomainError("forbidden", 403, "Source does not match service token")
    item = await session.scalar(
        select(Notification)
        .where(
            Notification.source == source.code,
            Notification.external_id == body.external_id,
        )
        .with_for_update()
    )
    if item is None:
        raise DomainError("not_found", 404, "Resource was not found")
    changed = item.lifecycle_status != "closed"
    if changed:
        item.lifecycle_status = "closed"
        item.close_reason = body.close_reason
        item.closed_at = datetime.now(UTC)
        session.add(
            audit(
                "notification.cancelled",
                context,
                None,
                "notification",
                item.id,
                metadata={"source": source.code, "close_reason": body.close_reason},
            )
        )
        await session.commit()
    result = await notification_dto(session, item)
    if changed:
        await fanout.publish_user(
            item.user_id,
            {
                "type": "notification.closed",
                "occurred_at": datetime.now(UTC).isoformat(),
                "notification_id": str(item.id),
                "close_reason": item.close_reason,
                "unread_count": await unread_count(session, item.user_id, 15),
            },
        )
    return result


async def catalog(session: AsyncSession) -> list[dict[str, Any]]:
    rows = (
        (
            await session.execute(
                select(NotificationType).where(NotificationType.record_status == "A")
            )
        )
        .scalars()
        .all()
    )
    button_codes = {
        code
        for item in rows
        for code in (item.button_primary_code, item.button_secondary_code)
        if code
    }
    buttons = {
        item.code: item
        for item in (
            (
                await session.execute(
                    select(NotificationButton).where(
                        NotificationButton.code.in_(button_codes),
                        NotificationButton.record_status == "A",
                    )
                )
            )
            .scalars()
            .all()
            if button_codes
            else []
        )
    }

    def button(code: str | None) -> dict[str, str] | None:
        item = buttons.get(code or "")
        return {"code": item.code, "label": item.label} if item else None

    return [
        {
            "code": item.code,
            "label": item.label,
            "color_token": item.color_token,
            "icon_code": item.icon_code,
            "cta_text": item.cta_text,
            "cta_action": item.cta_action,
            "countable": item.countable,
            "contour": item.contour,
            "button_primary": button(item.button_primary_code),
            "button_secondary": button(item.button_secondary_code),
        }
        for item in sorted(rows, key=lambda value: (value.contour, value.priority, value.code))
    ]


async def public_notifications(session: AsyncSession, limit: int) -> list[dict[str, Any]]:
    now = datetime.now(UTC)
    rows = (
        await session.execute(
            select(GuestNotification, NotificationType)
            .join(
                NotificationType,
                NotificationType.code == GuestNotification.notification_type,
            )
            .where(
                GuestNotification.record_status == "A",
                GuestNotification.lifecycle_status == "active",
                (GuestNotification.date_expired.is_(None)) | (GuestNotification.date_expired > now),
                NotificationType.record_status == "A",
                NotificationType.contour == "G",
            )
            .order_by(
                func.coalesce(GuestNotification.priority_override, NotificationType.priority),
                GuestNotification.notification_datetime.desc(),
                GuestNotification.id.desc(),
            )
            .limit(limit)
        )
    ).all()
    return [
        {
            "id": item.id,
            "notification_type": item.notification_type,
            "notification_datetime": item.notification_datetime,
            "header": item.header,
            "text": item.text,
            "price": item.price,
            "old_price": item.old_price,
            "instruction_url": item.instruction_url,
            "instruction_open_mode": "new_tab" if item.instruction_url else None,
            "chat_message_text": item.chat_message_text,
        }
        for item, _kind in rows
    ]


def _active_notification_query(user_id: uuid.UUID):
    now = datetime.now(UTC)
    return (
        select(Notification, NotificationType)
        .join(NotificationType, NotificationType.code == Notification.notification_type)
        .where(
            Notification.user_id == user_id,
            Notification.record_status == "A",
            Notification.lifecycle_status == "active",
            (Notification.date_expired.is_(None)) | (Notification.date_expired > now),
            NotificationType.contour == "P",
        )
    )


async def list_notifications(
    session: AsyncSession, user_id: uuid.UUID, place: str, limit: int
) -> list[dict[str, Any]]:
    query = _active_notification_query(user_id)
    priority = func.coalesce(Notification.priority_override, NotificationType.priority)
    if place == "home":
        query = query.where(Notification.visibility == "visible").order_by(
            priority, Notification.notification_datetime.desc(), Notification.id.desc()
        )
    else:
        query = query.order_by(
            priority,
            case((Notification.is_read.is_(False), 0), else_=1),
            Notification.notification_datetime.desc(),
            Notification.id.desc(),
        )
    rows = (await session.execute(query.limit(limit))).all()
    return [await notification_dto(session, item, kind) for item, kind in rows]


async def unread_count(session: AsyncSession, user_id: uuid.UUID, limit: int) -> int:
    rows = await list_notifications(session, user_id, "center", limit)
    return sum(1 for item in rows if item["countable"] and not item["is_read"])


async def owned_notification(
    session: AsyncSession,
    user_id: uuid.UUID,
    notification_id: uuid.UUID,
    *,
    action: bool = False,
) -> tuple[Notification, NotificationType]:
    statement = (
        select(Notification, NotificationType)
        .join(NotificationType, NotificationType.code == Notification.notification_type)
        .where(
            Notification.id == notification_id,
            Notification.user_id == user_id,
            Notification.record_status == "A",
        )
    )
    if action:
        statement = statement.with_for_update()
    row = (await session.execute(statement)).one_or_none()
    if row is None:
        raise DomainError("not_found", 404, "Resource was not found")
    item, kind = row
    if item.lifecycle_status == "closed" or (
        item.date_expired is not None and item.date_expired <= datetime.now(UTC)
    ):
        if action:
            raise DomainError("notification_closed", 409, "Notification is closed")
        raise DomainError("not_found", 404, "Resource was not found")
    return item, kind


async def notification_dto(
    session: AsyncSession, item: Notification, kind: NotificationType | None = None
) -> dict[str, Any]:
    kind = kind or await _type(session, item.notification_type)
    details = dict(item.details or {})
    links = (
        await session.execute(
            select(NotificationDocument, Document)
            .join(Document, Document.id == NotificationDocument.document_id)
            .where(
                NotificationDocument.notification_id == item.id,
                NotificationDocument.record_status == "A",
                Document.record_status == "A",
            )
            .order_by(NotificationDocument.sort_order, NotificationDocument.id)
        )
    ).all()
    if links:
        details["documents"] = [
            {
                "document_id": document.id,
                "title": document.name,
                "mime_type": document.mime_type,
                "size_bytes": document.size_bytes,
            }
            for _link, document in links
        ]
    if details.get("send_documents"):
        drafts = (
            (
                await session.execute(
                    select(ClientUploadDraft).where(
                        ClientUploadDraft.user_id == item.user_id,
                        ClientUploadDraft.context_type == "notification",
                        ClientUploadDraft.context_id == item.id,
                        ClientUploadDraft.state == "draft",
                    )
                )
            )
            .scalars()
            .all()
        )
        details["pending_documents"] = [upload_dto(draft) for draft in drafts]
    return {
        "id": item.id,
        "notification_type": item.notification_type,
        "notification_datetime": item.notification_datetime,
        "header": item.header,
        "text": item.text,
        "priority": item.priority_override if item.priority_override is not None else kind.priority,
        "date_expired": item.date_expired,
        "price": item.price,
        "old_price": item.old_price,
        "details": details or None,
        "lifecycle_status": item.lifecycle_status,
        "visibility": item.visibility,
        "is_read": item.is_read,
        "close_reason": item.close_reason,
        "countable": kind.countable,
        "cta_action": kind.cta_action,
    }


def _apply_hidden_ttl(
    item: Notification, kind: NotificationType, snapshot: SettingsSnapshot
) -> None:
    item.visibility = "hidden"
    if item.date_expired is None:
        days = kind.hidden_ttl_days or snapshot.integer("notification.hidden.default_ttl_days")
        item.date_expired = datetime.now(UTC) + timedelta(days=days)


async def state_response(
    session: AsyncSession, item: Notification, center_limit: int, result: Any = None
) -> dict[str, Any]:
    return {
        "notification_id": item.id,
        "lifecycle_status": item.lifecycle_status,
        "visibility": item.visibility,
        "is_read": item.is_read,
        "close_reason": item.close_reason,
        "date_expired": item.date_expired,
        "unread_count": await unread_count(session, item.user_id, center_limit),
        "result": result,
    }


async def apply_read_or_hide(
    session: AsyncSession,
    user_id: uuid.UUID,
    notification_id: uuid.UUID,
    operation: str,
    snapshot: SettingsSnapshot,
    fanout: RealtimeFanout,
    context: AuditContext,
) -> dict[str, Any]:
    item, kind = await owned_notification(session, user_id, notification_id, action=True)
    center_limit = snapshot.integer("notification.center.max_items")
    changed: dict[str, Any] = {}
    if operation == "read" and not item.is_read:
        item.is_read = True
        changed["is_read"] = True
    if operation == "hide" and item.visibility != "hidden":
        _apply_hidden_ttl(item, kind, snapshot)
        changed["visibility"] = "hidden"
        changed["date_expired"] = item.date_expired
    if changed:
        session.add(
            audit(
                f"notification.{operation}",
                context,
                user_id,
                "notification",
                item.id,
            )
        )
        await session.commit()
        changed["unread_count"] = await unread_count(session, user_id, center_limit)
        await fanout.publish_user(
            user_id,
            {
                "type": "notification.updated",
                "occurred_at": datetime.now(UTC).isoformat(),
                "notification_id": str(item.id),
                **changed,
            },
        )
    return await state_response(session, item, center_limit)


async def press_button(
    session: AsyncSession,
    user_id: uuid.UUID,
    notification_id: uuid.UUID,
    code: str,
    snapshot: SettingsSnapshot,
    fanout: RealtimeFanout,
    context: AuditContext,
) -> dict[str, Any]:
    item, kind = await owned_notification(session, user_id, notification_id, action=True)
    if code not in {kind.button_primary_code, kind.button_secondary_code}:
        raise DomainError("button_not_allowed", 422, "Button is not allowed")
    button = await session.scalar(
        select(NotificationButton).where(
            NotificationButton.code == code, NotificationButton.record_status == "A"
        )
    )
    if button is None:
        raise DomainError("button_not_allowed", 422, "Button is not allowed")
    submission_id: uuid.UUID | None = None
    submitted = 0
    if button.submits_documents:
        drafts = (
            (
                await session.execute(
                    select(ClientUploadDraft)
                    .where(
                        ClientUploadDraft.user_id == user_id,
                        ClientUploadDraft.context_type == "notification",
                        ClientUploadDraft.context_id == item.id,
                        ClientUploadDraft.state == "draft",
                        ClientUploadDraft.scan_status == "clean",
                    )
                    .with_for_update()
                )
            )
            .scalars()
            .all()
        )
        if not drafts:
            raise DomainError("validation_error", 400, "At least one clean document is required")
        submission_id = uuid7()
        for draft in drafts:
            session.add(
                ClientDocument(
                    id=uuid7(),
                    user_id=user_id,
                    context_type=draft.context_type,
                    context_id=draft.context_id,
                    submission_id=submission_id,
                    source_draft_id=draft.id,
                    original_file_name=draft.original_file_name,
                    safe_file_name=draft.safe_file_name,
                    mime_type=draft.mime_type,
                    size_bytes=draft.size_bytes,
                    checksum_sha256=draft.checksum_sha256 or "",
                    storage_bucket=draft.storage_bucket,
                    object_key=draft.object_key,
                    submitted_at=datetime.now(UTC),
                )
            )
            draft.state = "submitted"
            draft.submission_id = submission_id
        submitted = len(drafts)
    if button.sets_hidden:
        if button.applies_hidden_ttl:
            _apply_hidden_ttl(item, kind, snapshot)
        else:
            item.visibility = "hidden"
    if button.close_reason:
        item.lifecycle_status = "closed"
        item.close_reason = button.close_reason
        item.closed_at = datetime.now(UTC)
    session.add(
        audit(
            "notification.button_pressed",
            context,
            user_id,
            "notification",
            item.id,
            metadata={
                "button_code": code,
                "submission_id": str(submission_id) if submission_id else None,
                "submitted_count": submitted,
            },
        )
    )
    await session.commit()
    unread = await unread_count(session, user_id, snapshot.integer("notification.center.max_items"))
    event_type = (
        "notification.closed" if item.lifecycle_status == "closed" else "notification.updated"
    )
    await fanout.publish_user(
        user_id,
        {
            "type": event_type,
            "occurred_at": datetime.now(UTC).isoformat(),
            "notification_id": str(item.id),
            "close_reason": item.close_reason,
            "visibility": item.visibility,
            "date_expired": item.date_expired,
            "unread_count": unread,
        },
    )
    return await state_response(session, item, snapshot.integer("notification.center.max_items"))


async def invoke_cta_state(
    session: AsyncSession,
    user_id: uuid.UUID,
    notification_id: uuid.UUID,
    snapshot: SettingsSnapshot,
    fanout: RealtimeFanout,
    context: AuditContext,
    *,
    chat_result: Any = None,
) -> tuple[Notification, NotificationType, dict[str, Any]]:
    item, kind = await owned_notification(session, user_id, notification_id, action=True)
    item.is_read = True
    result: dict[str, Any]
    if kind.cta_action == "open_detail":
        result = {"action": "open_detail", "notification_id": str(item.id)}
    elif kind.cta_action == "open_payment_url":
        result = {"action": "open_url", "url": item.payment_url}
    elif kind.cta_action == "send_chat_message":
        if chat_result is None:
            result = {"action": "send_chat_message", "text": item.chat_message_text}
        else:
            result = {"action": "chat_message_sent", "message": chat_result}
    else:
        raise DomainError("validation_error", 400, "CTA is unavailable for personal contour")
    if kind.cta_sets_hidden:
        item.visibility = "hidden"
    if kind.cta_close_reason:
        item.lifecycle_status = "closed"
        item.close_reason = kind.cta_close_reason
        item.closed_at = datetime.now(UTC)
    session.add(
        audit(
            "notification.cta_invoked",
            context,
            user_id,
            "notification",
            item.id,
            metadata={"cta_action": kind.cta_action},
        )
    )
    await session.commit()
    unread = await unread_count(session, user_id, snapshot.integer("notification.center.max_items"))
    await fanout.publish_user(
        user_id,
        {
            "type": (
                "notification.closed"
                if item.lifecycle_status == "closed"
                else "notification.updated"
            ),
            "occurred_at": datetime.now(UTC).isoformat(),
            "notification_id": str(item.id),
            "is_read": True,
            "visibility": item.visibility,
            "close_reason": item.close_reason,
            "unread_count": unread,
        },
    )
    return (
        item,
        kind,
        await state_response(
            session,
            item,
            snapshot.integer("notification.center.max_items"),
            result,
        ),
    )


async def document_download(
    session: AsyncSession,
    user_id: uuid.UUID,
    notification_id: uuid.UUID,
    document_id: uuid.UUID,
    snapshot: SettingsSnapshot,
    s3: S3Client,
    fanout: RealtimeFanout,
    context: AuditContext,
) -> dict[str, Any]:
    item, kind = await owned_notification(session, user_id, notification_id, action=True)
    row = (
        await session.execute(
            select(NotificationDocument, Document)
            .join(Document, Document.id == NotificationDocument.document_id)
            .where(
                NotificationDocument.notification_id == item.id,
                NotificationDocument.document_id == document_id,
                NotificationDocument.record_status == "A",
                Document.user_id == user_id,
                Document.record_status == "A",
            )
            .with_for_update()
        )
    ).one_or_none()
    if row is None:
        raise DomainError("not_found", 404, "Resource was not found")
    link, document = row
    first_download = (
        await session.scalar(
            select(func.count(NotificationDocument.id)).where(
                NotificationDocument.notification_id == item.id,
                NotificationDocument.download_url_issued_at.is_not(None),
            )
        )
        == 0
    )
    link.download_url_issued_at = link.download_url_issued_at or datetime.now(UTC)
    changed = False
    if first_download and kind.hide_on_document_download:
        item.is_read = True
        _apply_hidden_ttl(item, kind, snapshot)
        changed = True
    session.add(
        audit(
            "notification.document.download_url_issued",
            context,
            user_id,
            "document",
            document.id,
            metadata={"notification_id": str(item.id), "expires_in_seconds": 300},
        )
    )
    await session.commit()
    if changed:
        await fanout.publish_user(
            user_id,
            {
                "type": "notification.updated",
                "occurred_at": datetime.now(UTC).isoformat(),
                "notification_id": str(item.id),
                "is_read": True,
                "visibility": item.visibility,
                "date_expired": item.date_expired,
                "unread_count": await unread_count(
                    session, user_id, snapshot.integer("notification.center.max_items")
                ),
            },
        )
    return {
        "download_url": await s3.presign_get(document.storage_bucket, document.object_key),
        "expires_at": datetime.now(UTC) + timedelta(seconds=300),
    }


async def init_upload(
    session: AsyncSession,
    user_id: uuid.UUID,
    body: UploadInitRequest,
    snapshot: SettingsSnapshot,
    s3: S3Client,
) -> dict[str, Any]:
    item, _kind = await owned_notification(session, user_id, body.context_id)
    if not (item.details or {}).get("send_documents"):
        raise DomainError("validation_error", 400, "Notification does not accept documents")
    current = await session.scalar(
        select(func.count(ClientUploadDraft.id)).where(
            ClientUploadDraft.user_id == user_id,
            ClientUploadDraft.context_type == body.context_type,
            ClientUploadDraft.context_id == body.context_id,
            ClientUploadDraft.state == "draft",
        )
    )
    if int(current or 0) >= snapshot.integer("notification.documents.max_files"):
        raise DomainError("attachment_invalid", 400, "Document limit reached")
    extension = PurePath(body.file_name).suffix.lower().lstrip(".")
    if (
        extension not in snapshot.strings("chat.attachments.allowed_extensions")
        or extension in snapshot.strings("chat.attachments.disallowed_extensions")
        or body.mime_type not in snapshot.strings("chat.attachments.allowed_mime_types")
        or body.size_bytes > snapshot.integer("chat.attachments.max_size_mb") * 1024 * 1024
    ):
        raise DomainError("attachment_invalid", 400, "File type or size is not allowed")
    draft_id = uuid7()
    key = f"quarantine/users/{user_id}/uploads/{draft_id}"
    ttl = snapshot.integer("chat.attachments.presigned_upload_ttl_seconds")
    expires = datetime.now(UTC) + timedelta(seconds=ttl)
    safe_name = re.sub(r"[^A-Za-z0-9._-]", "_", unicodedata.normalize("NFKC", body.file_name))
    draft = ClientUploadDraft(
        id=draft_id,
        user_id=user_id,
        context_type=body.context_type,
        context_id=body.context_id,
        original_file_name=body.file_name,
        safe_file_name=safe_name,
        mime_type=body.mime_type,
        size_bytes=body.size_bytes,
        storage_bucket=s3.settings.selectel_s3_bucket_quarantine,
        object_key=key,
        quarantine_object_key=key,
        upload_expires_at=expires,
    )
    session.add(draft)
    await session.commit()
    return {
        "draft_id": draft.id,
        "upload_url": await s3.presign_put(key, body.mime_type, ttl),
        "upload_headers": {"Content-Type": body.mime_type},
        "expires_at": expires,
    }


async def complete_upload(
    session: AsyncSession,
    user_id: uuid.UUID,
    draft_id: uuid.UUID,
    body: UploadCompleteRequest,
    s3: S3Client,
    safety: SafetyClient,
    request_id: str,
) -> dict[str, Any]:
    draft = await _owned_draft(session, user_id, draft_id)
    checksum = body.checksum.removeprefix("sha256:")
    if draft.completed_at:
        if draft.checksum_sha256 != checksum:
            raise DomainError("resource_state_conflict", 409, "Checksum changed")
        return upload_dto(draft)
    try:
        metadata = await s3.head(draft.storage_bucket, draft.object_key)
    except Exception as exc:
        raise DomainError("dependency_unavailable", 503, "Object storage unavailable") from exc
    if (
        int(metadata["ContentLength"]) != draft.size_bytes
        or metadata.get("ContentType") != draft.mime_type
    ):
        raise DomainError("attachment_invalid", 400, "Uploaded metadata differs")
    version_id, etag = metadata.get("VersionId"), metadata.get("ETag")
    if not version_id or not etag:
        raise DomainError(
            "dependency_unavailable", 503, "Versioned object metadata is unavailable"
        )
    draft.checksum_sha256 = checksum
    draft.quarantine_version_id = str(version_id)
    draft.quarantine_etag = str(etag)
    verdict = await safety.check(
        {
            "message_id": str(draft.id),
            "content_kind": "file",
            "text": "",
            "attachment": {
                "attachment_id": str(draft.id),
                "quarantine_object_key": draft.quarantine_object_key,
                "quarantine_version_id": draft.quarantine_version_id,
                "quarantine_etag": draft.quarantine_etag,
                "checksum": body.checksum,
                "mime_type": draft.mime_type,
                "size_bytes": draft.size_bytes,
            },
        },
        request_id,
    )
    if verdict["_status"] == 202:
        deadline = datetime.now(UTC) + timedelta(
            seconds=safety.settings.message_safety_task_poll_max_sec
        )
        while verdict["_status"] == 202 and datetime.now(UTC) < deadline:
            await asyncio.sleep(safety.settings.message_safety_task_poll_interval_sec)
            verdict = await safety.poll(verdict["_location"], request_id)
    draft.safety_processing_mode = verdict.get("processing_mode")
    draft.safety_config_version = verdict.get("config_version")
    draft.safety_rules_version = verdict.get("rules_version")
    if verdict["_status"] == 200:
        destination = (
            f"attachments/users/{user_id}/{draft.context_type}/{draft.context_id}/{draft.id}"
        )
        await s3.promote(
            draft.quarantine_object_key or draft.object_key,
            destination,
            version_id=draft.quarantine_version_id or "",
            etag=draft.quarantine_etag or "",
        )
        draft.storage_bucket = s3.settings.selectel_s3_bucket_attachments
        draft.object_key = destination
        draft.quarantine_object_key = None
        draft.scan_status = (
            "bypassed" if verdict["processing_mode"] == "mock" else "clean"
        )
    elif verdict["_status"] == 403:
        if draft.quarantine_object_key:
            await s3.delete_quarantine(draft.quarantine_object_key)
        draft.scan_status = "infected"
    else:
        draft.scan_status = "failed"
    draft.completed_at = datetime.now(UTC)
    await session.commit()
    return upload_dto(draft)


async def _owned_draft(
    session: AsyncSession, user_id: uuid.UUID, draft_id: uuid.UUID
) -> ClientUploadDraft:
    draft = await session.scalar(
        select(ClientUploadDraft).where(
            ClientUploadDraft.id == draft_id,
            ClientUploadDraft.user_id == user_id,
        )
    )
    if draft is None:
        raise DomainError("not_found", 404, "Resource was not found")
    return draft


def upload_dto(draft: ClientUploadDraft) -> dict[str, Any]:
    return {
        "draft_id": draft.id,
        "context_type": draft.context_type,
        "context_id": draft.context_id,
        "title": draft.safe_file_name,
        "mime_type": draft.mime_type,
        "size_bytes": draft.size_bytes,
        "scan_status": draft.scan_status,
        "state": draft.state,
    }


async def list_uploads(
    session: AsyncSession, user_id: uuid.UUID, context_type: str, context_id: uuid.UUID
) -> list[dict[str, Any]]:
    await owned_notification(session, user_id, context_id)
    drafts = (
        (
            await session.execute(
                select(ClientUploadDraft)
                .where(
                    ClientUploadDraft.user_id == user_id,
                    ClientUploadDraft.context_type == context_type,
                    ClientUploadDraft.context_id == context_id,
                    ClientUploadDraft.state == "draft",
                )
                .order_by(ClientUploadDraft.created_at, ClientUploadDraft.id)
            )
        )
        .scalars()
        .all()
    )
    return [upload_dto(item) for item in drafts]


async def discard_upload(
    session: AsyncSession, user_id: uuid.UUID, draft_id: uuid.UUID, s3: S3Client
) -> None:
    draft = await _owned_draft(session, user_id, draft_id)
    if draft.state == "submitted":
        raise DomainError("resource_state_conflict", 409, "Document is already submitted")
    if draft.state != "discarded":
        if draft.quarantine_object_key:
            await s3.delete_quarantine(draft.quarantine_object_key)
        elif draft.storage_bucket and draft.object_key:
            await s3.delete(draft.storage_bucket, draft.object_key)
        draft.state = "discarded"
        await session.commit()


async def expire_notifications(session: AsyncSession) -> tuple[int, int]:
    now = datetime.now(UTC)
    locked = await session.scalar(select(func.pg_try_advisory_xact_lock(0x48414E4E4F544946)))
    if not locked:
        return 0, 0
    personal = await session.execute(
        update(Notification)
        .where(
            Notification.record_status == "A",
            Notification.lifecycle_status == "active",
            Notification.date_expired <= now,
        )
        .values(lifecycle_status="closed", close_reason="expired", closed_at=now)
    )
    guest = await session.execute(
        update(GuestNotification)
        .where(
            GuestNotification.record_status == "A",
            GuestNotification.lifecycle_status == "active",
            GuestNotification.date_expired <= now,
        )
        .values(lifecycle_status="closed", closed_at=now)
    )
    personal_count = int(getattr(personal, "rowcount", 0) or 0)
    guest_count = int(getattr(guest, "rowcount", 0) or 0)
    session.add(
        audit(
            "notification.expired_batch",
            AuditContext(
                request_id=f"notification-expire-{uuid7()}",
                trace_id=str(uuid7()),
                ux_session_id=None,
                user_agent_hash=None,
                client_ip=None,
            ),
            None,
            metadata={
                "personal_count": personal_count,
                "guest_count": guest_count,
            },
        )
    )
    await session.commit()
    return personal_count, guest_count