GallerySubscriber/backend/app/tasks/downloads.py

"""Download-related Celery tasks."""

import asyncio
import logging
import re
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Optional

from sqlalchemy import select, and_, or_, update
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload

from app.tasks.db import get_async_session, cleanup_engine as _cleanup_engine
from app.tasks.celery_app import celery_app
from app.models.base import utcnow
from app.models.source import Source
from app.models.subscription import Subscription
from app.models.download import Download, DownloadStatus, ErrorType as DBErrorType
from app.models.setting import Setting, DEFAULT_SETTINGS
from app.services.gallery_dl import GalleryDLService, SourceConfig, DownloadResult
from app.services.credential_manager import CredentialManager
from app.services.patreon_resolver import resolve_campaign_id
from app.events import publish_event

logger = logging.getLogger(__name__)

# Settings for database connection
from app.config import get_settings
settings = get_settings()


async def _get_db_setting(session: AsyncSession, key: str, default=None):
    """Get a setting value from the database, falling back to defaults.

    Args:
        session: Database session
        key: Setting key (e.g., 'download.rate_limit')
        default: Fallback if not in database or DEFAULT_SETTINGS

    Returns:
        The setting value
    """
    result = await session.execute(
        select(Setting).where(Setting.key == key)
    )
    setting = result.scalar()

    if setting:
        logger.debug(f"DB setting '{key}' found: {setting.value} (type: {type(setting.value).__name__})")
        return setting.value

    # Fall back to DEFAULT_SETTINGS, then to provided default
    fallback = DEFAULT_SETTINGS.get(key, default)
    logger.debug(f"DB setting '{key}' not found, using fallback: {fallback}")
    return fallback


def _get_db_setting_sync(key: str, default=None):
    """Synchronously get a setting value from the database.

    This is used for getting settings in Celery task retry logic.
    Creates a fresh connection for each call.
    """
    async def _fetch():
        session_factory, engine = get_async_session()
        try:
            async with session_factory() as session:
                return await _get_db_setting(session, key, default)
        finally:
            await _cleanup_engine(engine)

    return asyncio.run(_fetch())


# --- Patreon campaign-ID auto-resolution helpers -------------------------
# These support auto-healing gallery-dl's "Failed to extract campaign ID"
# failures on Patreon sources. See docs/superpowers/specs/2026-04-18-patreon-campaign-resolver-design.md.

_PATREON_VANITY_RE = re.compile(
    r"^https?://(?:www\.)?patreon\.com/(?:c/)?(?!id:)([^/?#]+)",
    re.IGNORECASE,
)
_CAMPAIGN_ID_FAILURE_PATTERN = "failed to extract campaign id"


def _effective_patreon_url(platform: str, source_url: str, metadata: Optional[dict]) -> str:
    """Return the URL to pass to gallery-dl for this source.

    For Patreon sources with a cached campaign_id in metadata, returns
    patreon.com/id:<id> to bypass the broken vanity-URL lookup path.
    All other cases return source_url unchanged.
    """
    if platform == "patreon" and metadata:
        campaign_id = metadata.get("patreon_campaign_id")
        if campaign_id:
            return f"https://www.patreon.com/id:{campaign_id}"
    return source_url


def _looks_like_campaign_id_failure(dl_result) -> bool:
    """True if the gallery-dl output contains the campaign-ID extraction failure marker."""
    stdout = getattr(dl_result, "stdout", None) or ""
    stderr = getattr(dl_result, "stderr", None) or ""
    return _CAMPAIGN_ID_FAILURE_PATTERN in f"{stdout}\n{stderr}".lower()


def _extract_vanity_from_url(url: str) -> Optional[str]:
    """Return the vanity (creator slug) from a Patreon URL, or None if not extractable.

    Handles patreon.com/<vanity>, patreon.com/c/<vanity>[/...], with or without www.
    Returns None for id:<N> URLs, bare domain, and non-Patreon URLs.
    """
    m = _PATREON_VANITY_RE.match(url)
    return m.group(1) if m else None


@dataclass
class _DownloadAttempt:
    """Result of executing a download, possibly with a retry after campaign-ID resolution."""
    dl_result: DownloadResult
    resolved_campaign_id: Optional[str] = None


async def _execute_download_with_retry(
    *,
    gdl_service,
    platform: str,
    source_url: str,
    source_metadata: dict,
    subscription_name: str,
    source_config,
    cookies_path: Optional[str],
    auth_token: Optional[str],
) -> _DownloadAttempt:
    """Run the download once, auto-heal Patreon campaign-ID failures with one retry.

    Returns a _DownloadAttempt wrapping:
      - the final DownloadResult (retry's result if we retried, else the first)
      - resolved_campaign_id: the newly-resolved ID (if we resolved one this call), else None

    The caller is responsible for persisting resolved_campaign_id into source.metadata_.
    """
    effective_url = _effective_patreon_url(platform, source_url, source_metadata)
    dl_result = await gdl_service.download(
        url=effective_url,
        subscription_name=subscription_name,
        platform=platform,
        source_config=source_config,
        cookies_path=cookies_path,
        auth_token=auth_token,
    )

    # Only retry if: patreon + campaign-ID failure + no cached id + extractable vanity
    if (
        platform == "patreon"
        and not dl_result.success
        and "patreon_campaign_id" not in (source_metadata or {})
        and _looks_like_campaign_id_failure(dl_result)
    ):
        vanity = _extract_vanity_from_url(source_url)
        if vanity:
            logger.info(
                f"Attempting campaign-ID resolution for {subscription_name}/patreon ({vanity})"
            )
            campaign_id = await resolve_campaign_id(vanity, cookies_path)
            if campaign_id:
                retry_result = await gdl_service.download(
                    url=f"https://www.patreon.com/id:{campaign_id}",
                    subscription_name=subscription_name,
                    platform=platform,
                    source_config=source_config,
                    cookies_path=cookies_path,
                    auth_token=auth_token,
                )
                return _DownloadAttempt(
                    dl_result=retry_result, resolved_campaign_id=campaign_id
                )
            else:
                logger.warning(
                    f"Could not resolve campaign ID for {subscription_name}/patreon ({vanity}); "
                    f"original failure will be recorded"
                )

    return _DownloadAttempt(dl_result=dl_result, resolved_campaign_id=None)


async def _download_source_async(source_id: int, download_id: int = None) -> dict:
    """Async implementation of source download.

    Uses separate DB sessions for setup and result writing to avoid
    connection timeouts during long-running gallery-dl downloads.

    Args:
        source_id: ID of the source to download
        download_id: Optional ID of a pre-created Download record (QUEUED state)

    Returns:
        Dict with download results
    """
    # --- Phase 1: Setup (read source info, mark as RUNNING, get credentials) ---
    session_factory, engine = get_async_session()

    try:
        async with session_factory() as session:
            # Get source with subscription
            result = await session.execute(
                select(Source)
                .options(selectinload(Source.subscription))
                .where(Source.id == source_id)
            )
            source = result.scalar()

            if not source:
                logger.error(f"Source {source_id} not found")
                return {"source_id": source_id, "status": "error", "error": "Source not found"}

            if not source.enabled:
                logger.info(f"Source {source.subscription.name}/{source.platform} is disabled, skipping")
                if download_id:
                    dl_result = await session.execute(
                        select(Download).where(Download.id == download_id)
                    )
                    dl = dl_result.scalar()
                    if dl:
                        dl.status = DownloadStatus.SKIPPED
                        await session.commit()
                return {"source_id": source_id, "status": "skipped", "reason": "disabled"}

            # Capture values we need after the session closes
            subscription_name = source.subscription.name
            source_url = source.url
            source_platform = source.platform
            source_metadata = dict(source.metadata_ or {})
            logger.info(f"Starting download for {subscription_name}/{source_platform}")

            # Use existing download record or create a new one
            if download_id:
                result = await session.execute(
                    select(Download).where(Download.id == download_id)
                )
                download = result.scalar()
                if download:
                    # Guard against race conditions: only proceed if still QUEUED
                    if download.status != DownloadStatus.QUEUED:
                        logger.info(f"Download {download_id} is no longer QUEUED (status: {download.status}), skipping")
                        return {"source_id": source_id, "download_id": download_id, "status": "skipped", "reason": "already_processed"}

                    # Also check if source already has another RUNNING download
                    running_check = await session.execute(
                        select(Download.id).where(
                            Download.source_id == source_id,
                            Download.status == DownloadStatus.RUNNING,
                            Download.id != download_id
                        )
                    )
                    running_download_id = running_check.scalar()
                    if running_download_id:
                        # Mark this download as skipped since another is already running
                        download.status = DownloadStatus.SKIPPED
                        download.error_message = f"Skipped - another download ({running_download_id}) already running for this source"
                        await session.commit()
                        logger.info(f"Source {source_id} already has a running download, marked {download_id} as SKIPPED")
                        return {"source_id": source_id, "download_id": download_id, "status": "skipped", "reason": "source_already_running"}

                    download.status = DownloadStatus.RUNNING
                    download.started_at = utcnow()
                    download.completed_at = None
                else:
                    logger.warning(f"Download {download_id} not found, creating new record")
                    download = None

            if not download_id or download is None:
                # Check if source already has an active download before creating new one
                active_check = await session.execute(
                    select(Download.id).where(
                        Download.source_id == source_id,
                        Download.status.in_([DownloadStatus.QUEUED, DownloadStatus.RUNNING])
                    )
                )
                if active_check.scalar():
                    logger.info(f"Source {source_id} already has an active download, skipping new download creation")
                    return {"source_id": source_id, "status": "skipped", "reason": "source_already_active"}

                download = Download(
                    source_id=source_id,
                    url=source_url,
                    status=DownloadStatus.RUNNING,
                    started_at=utcnow(),
                )
                session.add(download)

            await session.commit()
            await session.refresh(download)
            actual_download_id = download.id

            # Broadcast download started event
            publish_event("download.started", {
                "download_id": actual_download_id,
                "source_id": source_id,
                "subscription_name": subscription_name,
                "platform": source_platform,
                "url": source_url,
            })

            # Get credentials before closing session
            cred_manager = CredentialManager(session)
            cookies_path = None
            auth_token = None

            if source_platform in ("discord", "pixiv"):
                auth_token = await cred_manager.get_token(source_platform)
                if not auth_token:
                    logger.warning(f"No token for {source_platform}, download may fail")
            else:
                cookies_path = await cred_manager.get_cookies_path(source_platform)
                if not cookies_path and source_platform in ["patreon", "subscribestar"]:
                    logger.warning(f"No credentials for {source_platform}, download may fail")

            # Get rate limit from database settings
            db_rate_limit = await _get_db_setting(session, "download.rate_limit", settings.download_rate_limit)
            db_validate_files = bool(
                await _get_db_setting(session, "download.validate_files", True)
            )

        # Session is now closed - DB connection released
    finally:
        await _cleanup_engine(engine)

    # --- Phase 2: Execute download (no DB connection held) ---
    source_config = SourceConfig.from_dict(source_metadata)
    dl_result = None
    dl_error = None
    resolved_campaign_id: Optional[str] = None

    try:
        gdl_service = GalleryDLService(
            rate_limit=db_rate_limit,
            validate_files=db_validate_files,
        )
        attempt = await _execute_download_with_retry(
            gdl_service=gdl_service,
            platform=source_platform,
            source_url=source_url,
            source_metadata=source_metadata,
            subscription_name=subscription_name,
            source_config=source_config,
            cookies_path=cookies_path,
            auth_token=auth_token,
        )
        dl_result = attempt.dl_result
        resolved_campaign_id = attempt.resolved_campaign_id
    except Exception as e:
        logger.exception(f"Unexpected error downloading {subscription_name}/{source_platform}: {e}")
        dl_error = e

    # --- Phase 3: Write results (fresh DB session) ---
    session_factory2, engine2 = get_async_session()

    try:
        async with session_factory2() as session:
            # Re-fetch download and source with fresh session
            download = (await session.execute(
                select(Download).where(Download.id == actual_download_id)
            )).scalar()
            source = (await session.execute(
                select(Source).where(Source.id == source_id)
            )).scalar()

            if not download or not source:
                logger.error(f"Could not re-fetch download {actual_download_id} or source {source_id} for result update")
                return {"source_id": source_id, "status": "error", "error": "Lost DB records"}

            # Persist resolved campaign ID (if the retry hook resolved one this run)
            if resolved_campaign_id:
                source.metadata_ = {
                    **(source.metadata_ or {}),
                    "patreon_campaign_id": resolved_campaign_id,
                }

            if dl_error:
                download.status = DownloadStatus.FAILED
                download.error_type = DBErrorType.UNKNOWN_ERROR
                download.error_message = str(dl_error)
                download.completed_at = utcnow()
                source.error_count = (source.error_count or 0) + 1
                source.last_check = utcnow()
                await session.commit()

                publish_event("download.failed", {
                    "download_id": actual_download_id,
                    "source_id": source_id,
                    "subscription_name": subscription_name,
                    "platform": source_platform,
                    "error_type": "unknown_error",
                    "error_message": str(dl_error),
                })

                return {
                    "source_id": source_id,
                    "download_id": actual_download_id,
                    "status": "error",
                    "error": str(dl_error),
                }

            # Update download record with results
            download.completed_at = utcnow()
            download.file_count = dl_result.files_downloaded
            # Persist logs in three forms for the Downloads modal:
            #   stdout/stderr: raw streams, capped via _truncate_log so huge
            #     verbose runs don't balloon the JSONB row.
            #   stderr_errors_warnings: pre-filtered [error]/[warning] lines so
            #     the UI can show a concise errors-only panel by default.
            #   run_stats: structured counts for the modal's header row.
            run_stats = gdl_service._compute_run_stats(
                dl_result.return_code, dl_result.stdout, dl_result.stderr
            )
            run_stats["quarantined_count"] = dl_result.files_quarantined
            stderr_summary = gdl_service._extract_errors_warnings(dl_result.stderr)
            download.metadata_ = {
                "stdout": gdl_service._truncate_log(dl_result.stdout) or None,
                "stderr": gdl_service._truncate_log(dl_result.stderr) or None,
                "stderr_errors_warnings": stderr_summary or None,
                "run_stats": run_stats,
                "duration_seconds": dl_result.duration_seconds,
                "quarantined_paths": dl_result.quarantined_paths or None,
            }

            if dl_result.success:
                download.status = DownloadStatus.COMPLETED
                # Some success states still carry an informational error_type
                # (e.g., TIER_LIMITED — run completed, but the subscription tier
                # blocked posts we'd have otherwise downloaded). Persist it so
                # the UI can distinguish "fully succeeded" from "succeeded with
                # external constraints".
                download.error_type = dl_result.error_type.value if dl_result.error_type else None
                download.error_message = dl_result.error_message
                source.last_success = utcnow()
                source.error_count = 0
                logger.info(f"Download completed for {subscription_name}/{source_platform}: {dl_result.files_downloaded} files")

                # Mark any prior failures for this source as superseded
                await session.execute(
                    update(Download)
                    .where(
                        Download.source_id == source_id,
                        Download.status == DownloadStatus.FAILED,
                    )
                    .values(superseded=True)
                )

                publish_event("download.completed", {
                    "download_id": actual_download_id,
                    "source_id": source_id,
                    "subscription_name": subscription_name,
                    "platform": source_platform,
                    "file_count": dl_result.files_downloaded,
                    "duration_seconds": dl_result.duration_seconds,
                })
            else:
                download.status = DownloadStatus.FAILED
                download.error_type = dl_result.error_type.value if dl_result.error_type else None
                download.error_message = dl_result.error_message
                source.error_count = (source.error_count or 0) + 1
                logger.warning(f"Download failed for {subscription_name}/{source_platform}: {dl_result.error_message}")

                publish_event("download.failed", {
                    "download_id": actual_download_id,
                    "source_id": source_id,
                    "subscription_name": subscription_name,
                    "platform": source_platform,
                    "error_type": dl_result.error_type.value if dl_result.error_type else None,
                    "error_message": dl_result.error_message,
                })

            source.last_check = utcnow()
            await session.commit()

            return {
                "source_id": source_id,
                "download_id": actual_download_id,
                "status": "completed" if dl_result.success else "failed",
                "files_downloaded": dl_result.files_downloaded,
                "error_type": dl_result.error_type.value if dl_result.error_type else None,
                "error_message": dl_result.error_message,
                "duration_seconds": dl_result.duration_seconds,
            }
    finally:
        await _cleanup_engine(engine2)


async def _scheduled_check_async() -> dict:
    """Async implementation of scheduled source check.

    Returns:
        Dict with check results
    """
    session_factory, engine = get_async_session()

    try:
        async with session_factory() as session:
            now = utcnow()

            # Get global schedule interval from database settings
            # This is the primary interval that controls how often sources are checked
            raw_interval = await _get_db_setting(
                session, "download.schedule_interval", 28800  # Default 8 hours
            )
            # Ensure we have an integer value (JSONB might return different types)
            global_interval = int(raw_interval) if raw_interval is not None else 28800
            logger.info(f"Schedule check: using interval {global_interval}s (raw value: {raw_interval}, type: {type(raw_interval).__name__})")

            # Find sources due for checking:
            # - enabled = True
            # - last_check is NULL OR (now - last_check) > global_interval
            # Order by subscription priority, then by when last checked
            query = (
                select(Source)
                .options(selectinload(Source.subscription))
                .join(Subscription)
                .where(
                    and_(
                        Source.enabled == True,
                        Subscription.enabled == True,  # Subscription must also be enabled
                        or_(
                            Source.last_check == None,
                            Source.last_check < now - timedelta(seconds=1)  # Will be refined below
                        )
                    )
                )
                .order_by(
                    Subscription.priority.desc(),  # Higher subscription priority first
                    Source.last_check.asc().nullsfirst(),  # Never checked first
                )
            )

            result = await session.execute(query)
            sources = result.scalars().all()

            # Filter by global schedule interval from Settings
            # All sources use the same interval - no per-source overrides
            due_sources = []
            not_due_count = 0
            for source in sources:
                if source.last_check is None:
                    due_sources.append(source)
                    logger.debug(f"Source {source.id} ({source.platform}) - never checked, marking due")
                else:
                    time_since_check = (now - source.last_check).total_seconds()
                    if time_since_check >= global_interval:
                        due_sources.append(source)
                        logger.debug(f"Source {source.id} ({source.platform}) - last check {time_since_check:.0f}s ago >= {global_interval}s, marking due")
                    else:
                        not_due_count += 1
                        logger.debug(f"Source {source.id} ({source.platform}) - last check {time_since_check:.0f}s ago < {global_interval}s, not due yet")

            logger.info(f"Schedule check: {len(due_sources)} sources due, {not_due_count} sources not due (interval: {global_interval}s)")

            # Get source IDs that already have a queued or running download
            due_source_ids = [s.id for s in due_sources]
            active_result = await session.execute(
                select(Download.source_id).where(
                    Download.source_id.in_(due_source_ids),
                    Download.status.in_([DownloadStatus.QUEUED, DownloadStatus.RUNNING])
                )
            )
            active_source_ids = set(active_result.scalars().all())

            # Create download records and queue tasks for each eligible source
            queued = 0
            skipped = 0
            for source in due_sources:
                if source.id in active_source_ids:
                    skipped += 1
                    logger.debug(f"Skipping {source.subscription.name}/{source.platform} - already active")
                    continue

                download = Download(
                    source_id=source.id,
                    url=source.url,
                    status=DownloadStatus.QUEUED,
                )
                session.add(download)
                await session.commit()
                await session.refresh(download)

                download_source.delay(source.id, download_id=download.id)
                queued += 1
                logger.debug(f"Queued download for {source.subscription.name}/{source.platform}")

            if skipped:
                logger.info(f"Skipped {skipped} sources with active downloads")

            return {
                "checked": len(sources),
                "queued": queued,
                "timestamp": now.isoformat(),
            }
    finally:
        await _cleanup_engine(engine)


async def _cleanup_old_downloads_async() -> dict:
    """Async implementation of download cleanup.

    Returns:
        Dict with cleanup results
    """
    session_factory, engine = get_async_session()

    try:
        async with session_factory() as session:
            now = utcnow()

            # Delete completed downloads older than 30 days
            completed_cutoff = now - timedelta(days=30)
            completed_query = select(Download).where(
                and_(
                    Download.status == DownloadStatus.COMPLETED,
                    Download.created_at < completed_cutoff
                )
            )
            completed_result = await session.execute(completed_query)
            completed_old = completed_result.scalars().all()

            # Delete failed downloads older than 7 days
            failed_cutoff = now - timedelta(days=7)
            failed_query = select(Download).where(
                and_(
                    Download.status == DownloadStatus.FAILED,
                    Download.created_at < failed_cutoff
                )
            )
            failed_result = await session.execute(failed_query)
            failed_old = failed_result.scalars().all()

            deleted_count = 0
            for download in completed_old + failed_old:
                await session.delete(download)
                deleted_count += 1

            await session.commit()

            logger.info(f"Cleaned up {deleted_count} old download records")
            return {"deleted": deleted_count}
    finally:
        await _cleanup_engine(engine)


# Celery tasks that wrap async functions

@celery_app.task(bind=True)
def download_source(self, source_id: int, download_id: int = None):
    """Download new content from a single source.

    Args:
        source_id: ID of the source to download from
        download_id: Optional ID of a pre-created Download record (QUEUED state)
    """
    try:
        return asyncio.run(_download_source_async(source_id, download_id=download_id))
    except Exception as e:
        logger.exception(f"Task failed for source {source_id}: {e}")
        # Get retry count from database settings
        max_retries = _get_db_setting_sync("download.retry_count", 3)
        if self.request.retries < max_retries:
            # Retry with exponential backoff
            raise self.retry(exc=e, countdown=60 * (2 ** self.request.retries), max_retries=max_retries)
        else:
            logger.error(f"Max retries ({max_retries}) exceeded for source {source_id}")
            raise


@celery_app.task
def scheduled_check():
    """Check all enabled sources that are due for a check.

    This is called periodically by Celery Beat.
    """
    logger.info("=== SCHEDULED_CHECK TASK STARTING ===")
    try:
        result = asyncio.run(_scheduled_check_async())
        logger.info(f"=== SCHEDULED_CHECK TASK COMPLETE: {result} ===")
        return result
    except Exception as e:
        logger.exception(f"=== SCHEDULED_CHECK TASK FAILED: {e} ===")
        raise


@celery_app.task
def cleanup_old_downloads():
    """Clean up old download records.

    Keeps the database from growing indefinitely.
    """
    return asyncio.run(_cleanup_old_downloads_async())


@celery_app.task(bind=True)
def process_download(self, download_id: int):
    """Process a single download record (for retries).

    Args:
        download_id: ID of the download to process
    """
    async def _process():
        session_factory, engine = get_async_session()

        try:
            async with session_factory() as session:
                result = await session.execute(
                    select(Download).where(Download.id == download_id)
                )
                download = result.scalar()

                if not download:
                    return {"download_id": download_id, "status": "error", "error": "Download not found"}

                if not download.source_id:
                    return {"download_id": download_id, "status": "error", "error": "No source associated"}

                # Delegate to download_source
                return await _download_source_async(download.source_id, download_id=download_id)
        finally:
            await _cleanup_engine(engine)

    try:
        return asyncio.run(_process())
    except Exception as e:
        logger.exception(f"Task failed for download {download_id}: {e}")
        # Get retry count from database settings
        max_retries = _get_db_setting_sync("download.retry_count", 3)
        if self.request.retries < max_retries:
            raise self.retry(exc=e, countdown=60 * (2 ** self.request.retries), max_retries=max_retries)
        else:
            logger.error(f"Max retries ({max_retries}) exceeded for download {download_id}")
            raise