g2s-aggregator/app/services/waybill_email_sender.py

"""Background service that e-mails provider waybill PDFs to customers."""

import asyncio
from collections.abc import Callable, Mapping, Sequence
from contextlib import AbstractAsyncContextManager
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Protocol

import structlog

logger = structlog.get_logger(__name__)


_EMAIL_SUBJECT_TEMPLATE = "Накладная по заказу {order_uuid}"
_EMAIL_BODY_TEMPLATE = (
    "Здравствуйте!\n\n"
    "По вашему заказу {order_uuid} сформирована "
    "транспортная накладная.\n"
    "PDF-файл накладной приложен к этому письму.\n"
)
_EMAIL_BODY_URL_LINE_TEMPLATE = (
    "Также накладная доступна по ссылке: {waybill_url}\n"
)
_ATTACHMENT_FILENAME_TEMPLATE = "waybill_{order_uuid}.pdf"


class WaybillPDFDownloaderProtocol(Protocol):
    async def download_waybill_pdf(self, order: "OrderRecord") -> bytes: ...


class EmailSenderProtocol(Protocol):
    async def send_email(
        self,
        *,
        to: str,
        subject: str,
        body: str,
        attachment_bytes: bytes,
        attachment_filename: str,
    ) -> None: ...


class OrderRecord(Protocol):
    order_uuid: str
    provider: str
    account_email: str
    provider_waybill_id: str | None
    provider_waybill_url: str | None


class WaybillEmailSenderRepositoryProtocol(Protocol):
    def session(self) -> AbstractAsyncContextManager[object]: ...

    async def list_orders_pending_waybill_email(
        self, session: object, *, limit: int
    ) -> Sequence[OrderRecord]: ...

    async def record_waybill_email_sent(
        self,
        session: object,
        *,
        order_uuid: str,
        sent_at: datetime,
    ) -> object | None: ...


@dataclass(frozen=True)
class SendBatchSummary:
    processed: int
    succeeded: int
    failed: int


class WaybillEmailSenderService:
    def __init__(
        self,
        *,
        order_repository: WaybillEmailSenderRepositoryProtocol,
        waybill_downloaders: Mapping[str, WaybillPDFDownloaderProtocol],
        email_sender: EmailSenderProtocol,
        batch_size: int,
        datetime_now: Callable[[], datetime] = lambda: datetime.now(timezone.utc),
    ) -> None:
        self._repository = order_repository
        self._waybill_downloaders = dict(waybill_downloaders)
        self._email_sender = email_sender
        self._batch_size = batch_size
        self._datetime_now = datetime_now

    async def poll_once(self) -> SendBatchSummary:
        async with self._repository.session() as session:
            orders = await self._repository.list_orders_pending_waybill_email(
                session, limit=self._batch_size
            )
            succeeded = 0
            failed = 0
            for order in orders:
                try:
                    await self._handle_order(session, order)
                    succeeded += 1
                except Exception:
                    failed += 1
                    logger.exception(
                        "waybill_email_order_failed",
                        order_uuid=order.order_uuid,
                        provider=order.provider,
                        account_email=order.account_email,
                    )
            return SendBatchSummary(
                processed=len(orders),
                succeeded=succeeded,
                failed=failed,
            )

    async def run_forever(
        self,
        *,
        interval_seconds: float,
        stop_event: asyncio.Event,
    ) -> None:
        while not stop_event.is_set():
            try:
                summary = await self.poll_once()
                logger.debug(
                    "waybill_email_tick",
                    processed=summary.processed,
                    succeeded=summary.succeeded,
                    failed=summary.failed,
                )
            except Exception:
                logger.exception("waybill_email_tick_failed")

            try:
                await asyncio.wait_for(
                    stop_event.wait(), timeout=interval_seconds
                )
            except asyncio.TimeoutError:
                continue

    async def _handle_order(self, session: object, order: OrderRecord) -> None:
        if order.provider_waybill_url is None and order.provider_waybill_id is None:
            return

        downloader = self._resolve_downloader(order.provider)
        pdf_bytes = await downloader.download_waybill_pdf(order)
        subject = _EMAIL_SUBJECT_TEMPLATE.format(order_uuid=order.order_uuid)
        body = _EMAIL_BODY_TEMPLATE.format(order_uuid=order.order_uuid)
        if order.provider_waybill_url is not None:
            body += _EMAIL_BODY_URL_LINE_TEMPLATE.format(
                waybill_url=order.provider_waybill_url,
            )
        filename = _ATTACHMENT_FILENAME_TEMPLATE.format(order_uuid=order.order_uuid)

        await self._email_sender.send_email(
            to=order.account_email,
            subject=subject,
            body=body,
            attachment_bytes=pdf_bytes,
            attachment_filename=filename,
        )

        sent_at = self._datetime_now()
        await self._repository.record_waybill_email_sent(
            session,
            order_uuid=order.order_uuid,
            sent_at=sent_at,
        )
        logger.info(
            "waybill_email_sent",
            order_uuid=order.order_uuid,
            provider=order.provider,
            account_email=order.account_email,
            sent_at=sent_at,
        )

    def _resolve_downloader(self, provider: str) -> WaybillPDFDownloaderProtocol:
        downloader = self._waybill_downloaders.get(provider)
        if downloader is None:
            raise RuntimeError(f"Waybill downloader is not configured for {provider}.")
        return downloader