diff --git a/backend/app/api/downloads.py b/backend/app/api/downloads.py index 0a4e820..98fd48b 100644 --- a/backend/app/api/downloads.py +++ b/backend/app/api/downloads.py @@ -111,8 +111,8 @@ async def retry_download(download_id: int): if download.status != DownloadStatus.FAILED: return jsonify({"error": "Can only retry failed downloads"}), 400 - # Reset status to pending - download.status = DownloadStatus.PENDING + # Reset status to queued + download.status = DownloadStatus.QUEUED download.error_type = None download.error_message = None await session.commit() diff --git a/backend/app/api/sources.py b/backend/app/api/sources.py index 1f781bf..662bd89 100644 --- a/backend/app/api/sources.py +++ b/backend/app/api/sources.py @@ -5,8 +5,10 @@ from sqlalchemy import select, func from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload +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 from app.tasks.downloads import download_source from app.api.settings import get_setting_value @@ -216,13 +218,34 @@ async def trigger_check(source_id: int): if not source: return jsonify({"error": "Source not found"}), 404 + # Check for existing queued or running download for this source + existing = await session.execute( + select(Download).where( + Download.source_id == source.id, + Download.status.in_([DownloadStatus.QUEUED, DownloadStatus.RUNNING]) + ) + ) + if existing.scalar(): + return jsonify({"error": "A download is already queued or running for this source"}), 409 + + # Create download record in QUEUED state for visibility + download = Download( + source_id=source.id, + url=source.url, + status=DownloadStatus.QUEUED, + ) + session.add(download) + await session.commit() + await session.refresh(download) + # Queue Celery task for immediate download - task = download_source.delay(source_id) + task = download_source.delay(source_id, download_id=download.id) current_app.logger.info(f"Triggered check for source: {source.subscription.name}/{source.platform}") return jsonify({ "message": "Check queued", "source_id": source_id, + "download_id": download.id, "subscription_name": source.subscription.name, "platform": source.platform, "task_id": task.id, diff --git a/backend/app/api/subscriptions.py b/backend/app/api/subscriptions.py index 23ab742..b6bd8b6 100644 --- a/backend/app/api/subscriptions.py +++ b/backend/app/api/subscriptions.py @@ -7,6 +7,7 @@ from sqlalchemy.orm import selectinload from app.models.subscription import Subscription from app.models.source import Source +from app.models.download import Download, DownloadStatus from app.tasks.downloads import download_source from app.api.settings import get_setting_value @@ -303,16 +304,46 @@ async def trigger_subscription_check(subscription_id: int): enabled_sources = [s for s in subscription.sources if s.enabled] - # Queue Celery tasks for each enabled source - task_ids = [] - for source in enabled_sources: - task = download_source.delay(source.id) - task_ids.append(task.id) + # Get source IDs that already have a queued or running download + active_result = await session.execute( + select(Download.source_id).where( + Download.source_id.in_([s.id for s in enabled_sources]), + Download.status.in_([DownloadStatus.QUEUED, DownloadStatus.RUNNING]) + ) + ) + active_source_ids = set(active_result.scalars().all()) - current_app.logger.info(f"Triggered check for subscription: {subscription.name} ({len(enabled_sources)} sources)") + # Create download records and queue Celery tasks for each eligible source + task_ids = [] + download_ids = [] + skipped = 0 + for source in enabled_sources: + if source.id in active_source_ids: + skipped += 1 + continue + + download = Download( + source_id=source.id, + url=source.url, + status=DownloadStatus.QUEUED, + ) + session.add(download) + await session.commit() + await session.refresh(download) + + task = download_source.delay(source.id, download_id=download.id) + task_ids.append(task.id) + download_ids.append(download.id) + + current_app.logger.info( + f"Triggered check for subscription: {subscription.name} " + f"({len(download_ids)} queued, {skipped} already active)" + ) return jsonify({ "message": "Checks queued", "subscription_id": subscription_id, - "source_count": len(enabled_sources), + "source_count": len(download_ids), + "skipped_active": skipped, "task_ids": task_ids, + "download_ids": download_ids, }), 202 diff --git a/backend/app/models/download.py b/backend/app/models/download.py index 82231f4..ac52854 100644 --- a/backend/app/models/download.py +++ b/backend/app/models/download.py @@ -11,7 +11,8 @@ from app.models.base import Base, TimestampMixin, format_datetime class DownloadStatus: """Download status constants.""" - PENDING = "pending" + QUEUED = "queued" + PENDING = "pending" # Legacy: used by retry flow RUNNING = "running" COMPLETED = "completed" FAILED = "failed" diff --git a/backend/app/tasks/downloads.py b/backend/app/tasks/downloads.py index 6db390c..ffffe71 100644 --- a/backend/app/tasks/downloads.py +++ b/backend/app/tasks/downloads.py @@ -41,11 +41,12 @@ async def _cleanup_engine(engine): await engine.dispose() -async def _download_source_async(source_id: int) -> dict: +async def _download_source_async(source_id: int, download_id: int = None) -> dict: """Async implementation of source download. 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 @@ -68,19 +69,42 @@ async def _download_source_async(source_id: int) -> dict: if not source.enabled: logger.info(f"Source {source.subscription.name}/{source.platform} is disabled, skipping") + # If there's a queued download record, mark it as skipped + 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"} subscription_name = source.subscription.name logger.info(f"Starting download for {subscription_name}/{source.platform}") - # Create download record - download = Download( - source_id=source.id, - url=source.url, - status=DownloadStatus.RUNNING, - started_at=utcnow(), - ) - session.add(download) + # 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: + download.status = DownloadStatus.RUNNING + download.started_at = utcnow() + else: + logger.warning(f"Download {download_id} not found, creating new record") + download = None + + if not download_id or download is None: + 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) @@ -261,13 +285,41 @@ async def _scheduled_check_async() -> dict: logger.info(f"Found {len(due_sources)} sources due for checking") - # Queue download tasks for each source + # 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: - download_source.delay(source.id) + 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, @@ -327,14 +379,15 @@ async def _cleanup_old_downloads_async() -> dict: # Celery tasks that wrap async functions @celery_app.task(bind=True, max_retries=3) -def download_source(self, source_id: int): +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)) + 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}") # Retry with exponential backoff