"""User-facing video delivery, access checks and durable auto-deletion.""" from __future__ import annotations import logging import uuid from dataclasses import dataclass from datetime import timedelta from aiogram.exceptions import ( TelegramBadRequest, TelegramForbiddenError, TelegramNetworkError, TelegramRetryAfter, TelegramServerError, ) from app.config import Settings from app.database.base import utcnow from app.database.models import ( AccessToken, DeleteStatus, JobType, OperationStatus, Post, PostStatus, User, UserDelivery, ) from app.database.repositories import Repositories from app.errors import DeliveryError, RateLimitedError, RetryableJobError, TokenError from app.gateways.base import TelegramGateway from app.services.membership_service import MembershipResult, MembershipService from app.services.rate_limit_service import RateLimitService from app.services.settings_service import SettingsService from app.services.token_service import TokenService from app.utils.retry import classify_exception, sleep_before_retry logger = logging.getLogger(__name__) @dataclass(frozen=True, slots=True) class DeliveryOutcome: kind: str post: Post token: AccessToken membership: MembershipResult | None = None delivery: UserDelivery | None = None class DeliveryService: def __init__( self, repos: Repositories, settings: Settings, app_settings: SettingsService, gateway: TelegramGateway, tokens: TokenService, membership: MembershipService, rate_limit: RateLimitService, ) -> None: self.repos = repos self.settings = settings self.app_settings = app_settings self.gateway = gateway self.tokens = tokens self.membership = membership self.rate_limit = rate_limit async def request( self, raw_token: str, *, telegram_id: int, username: str | None, display_name: str | None, ) -> DeliveryOutcome: validation = await self.tokens.validate(raw_token) if not validation.valid or validation.post is None or validation.access_token is None: raise TokenError(_token_message(validation.reason)) post = validation.post token = validation.access_token if post.status in {PostStatus.ARCHIVED, PostStatus.PAUSED}: raise TokenError("این لینک در حال حاضر غیرفعال است.") if post.published_at is None and post.status != PostStatus.PUBLISHED: raise TokenError("این ویدیو هنوز منتشر نشده است.") user = await self.repos.users.get_or_create( telegram_id, username=username, display_name=display_name ) if user.is_blocked: raise DeliveryError("دسترسی شما به این ربات مسدود شده است.") try: await self.rate_limit.enforce( telegram_id, "delivery", max_requests=await self.app_settings.get_int( "rate_limit_requests", self.settings.delivery_rate_limit_requests ), window_seconds=await self.app_settings.get_int( "rate_limit_window_seconds", self.settings.delivery_rate_limit_window_seconds, ), min_interval_seconds=await self.app_settings.get_int( "rate_limit_min_interval_seconds", self.settings.delivery_min_interval_seconds, ), ) except RateLimitedError: raise membership_result = await self.membership.check_user(post.id, telegram_id) if not membership_result.all_joined: return DeliveryOutcome( "membership", post=post, token=token, membership=membership_result, ) delivery = await self._deliver(post, token, user) return DeliveryOutcome("delivered", post=post, token=token, delivery=delivery) async def _deliver(self, post: Post, token: AccessToken, user: User) -> UserDelivery: correlation_id = uuid.uuid4().hex if post.media_asset is None: # Do not burn a one-use token when delivery is impossible. raise DeliveryError("فایل ویدیوی این پست موجود نیست.") operation_id = await self.repos.operations.start( correlation_id, "deliver_user_video", entity_type="post", entity_id=post.id, meta={"user_id": user.telegram_id}, ) try: policy = await self.app_settings.delete_policy(post) protect = await self.app_settings.protect_content(post) cover = await self.app_settings.cover(post) buttons = await self.app_settings.buttons(post) channel_name = None destinations = await self.repos.posts.destinations(post.id) if destinations: channel_name = destinations[0].channel.title caption = await self.app_settings.build_caption(post, channel_name=channel_name) # Consume only after membership has passed and immediately before # the network send, so a configuration/rendering error does not # burn a one-use link. if not await self.repos.tokens.consume(token.id, now=utcnow()): await self.repos.operations.finish( operation_id, OperationStatus.FAILED, error_message="token already consumed", ) raise TokenError("ظرفیت استفاده از این لینک تکمیل شده است.") except TokenError: raise except Exception as exc: await self.repos.operations.finish( operation_id, OperationStatus.FAILED, error_message="delivery preparation failed", ) logger.warning( "delivery preparation failed for user %s: %s", user.telegram_id, type(exc).__name__, ) raise DeliveryError("ارسال ویدیو ناموفق بود. لطفاً بعداً دوباره تلاش کنید.") from exc try: message = await self._send_with_retry( user.telegram_id, post.media_asset.file_id, caption=caption, cover=cover, buttons=buttons, protect=protect, ) except Exception as exc: await self.repos.operations.finish( operation_id, OperationStatus.FAILED, error_message="delivery failed", ) logger.warning("delivery failed for user %s: %s", user.telegram_id, type(exc).__name__) raise DeliveryError("ارسال ویدیو ناموفق بود. لطفاً بعداً دوباره تلاش کنید.") from exc sent_at = utcnow() delete_at = ( sent_at + timedelta(seconds=policy.user_seconds) if policy.enabled and policy.user_seconds > 0 else None ) delivery = await self.repos.deliveries.create( user_id=user.id, post_id=post.id, access_token_id=token.id, telegram_message_id=message.message_id, sent_at=sent_at, delete_at=delete_at, correlation_id=correlation_id, ) if delete_at is not None: await self.repos.jobs.enqueue( job_type=JobType.DELETE_USER_MESSAGE, run_at=delete_at, payload={"delivery_id": delivery.id}, max_attempts=self.settings.job_max_attempts, correlation_id=correlation_id, idempotency_key=f"delete_delivery:{delivery.id}", post_id=post.id, delivery_id=delivery.id, ) await self.repos.operations.finish(operation_id, OperationStatus.SUCCESS) await self.repos.audit.log( user.telegram_id, "video.deliver", entity_type="post", entity_id=post.id, details={"delivery_id": delivery.id}, correlation_id=correlation_id, ) return delivery async def _send_with_retry( self, chat_id: int, file_id: str, *, caption: str, cover: str | None, buttons: list[tuple[str, str]], protect: bool, ): last_error: Exception | None = None for attempt in range(1, self.settings.retry_max_attempts + 1): try: return await self.gateway.send_video( chat_id, file_id, caption=caption, thumbnail_file_id=cover, protect_content=protect, buttons=buttons, ) except Exception as exc: last_error = exc decision = classify_exception(exc, sending=True) if isinstance(exc, (TelegramNetworkError, TelegramServerError, TimeoutError)): raise if isinstance(exc, TelegramRetryAfter) and attempt < self.settings.retry_max_attempts: await sleep_before_retry(attempt, decision, self.settings.retry_base_delay_seconds) continue if decision.retryable and attempt < self.settings.retry_max_attempts: await sleep_before_retry(attempt, decision, self.settings.retry_base_delay_seconds) continue raise if last_error: raise last_error raise DeliveryError("ارسال ویدیو ناموفق بود.") async def delete_user_message(self, delivery_id: int) -> None: delivery = await self.repos.deliveries.get(delivery_id) if delivery is None: raise RetryableJobError("رکورد ارسال پیدا نشد.", delay_seconds=60) if delivery.delete_status in {DeleteStatus.DELETED, DeleteStatus.SKIPPED}: return try: await self.gateway.delete_message( delivery.user.telegram_id, delivery.telegram_message_id ) except TelegramBadRequest as exc: text = str(exc).lower() if "not found" in text or "message to delete" in text: await self.repos.deliveries.mark_deleted(delivery_id) return await self.repos.deliveries.mark_deleted( delivery_id, error="حذف پیام با درخواست نامعتبر تلگرام مواجه شد." ) raise except TelegramRetryAfter as exc: raise RetryableJobError( "Telegram درخواست حذف را محدود کرد.", delay_seconds=exc.retry_after ) from exc except TelegramForbiddenError as exc: await self.repos.deliveries.mark_deleted( delivery_id, error="ربات اجازه حذف پیام را ندارد یا کاربر ربات را بلاک کرده است." ) raise RetryableJobError("حذف پیام مجاز نیست.", delay_seconds=3600) from exc except (TelegramNetworkError, TelegramServerError) as exc: raise RetryableJobError("خطای موقت در حذف پیام.", delay_seconds=60) from exc except Exception: await self.repos.deliveries.mark_deleted( delivery_id, error="حذف پیام ناموفق بود." ) raise await self.repos.deliveries.mark_deleted(delivery_id) await self.repos.audit.log( None, "user_message.delete", entity_type="user_delivery", entity_id=delivery_id, ) def _token_message(reason: str) -> str: return { "invalid": "لینک نامعتبر است.", "inactive": "این لینک غیرفعال شده است.", "expired": "مدت اعتبار این لینک به پایان رسیده است.", "exhausted": "ظرفیت استفاده از این لینک تکمیل شده است.", }.get(reason, "این لینک قابل استفاده نیست.")