"""Run storage ingest on the API process that can see the staged file.

arq often runs on a different host/container than the API. That worker must
not mark a job failed just because its local disk is empty — only the API
process that accepted the upload can push the bytes to R2.
"""
from __future__ import annotations

import asyncio
from datetime import datetime, timedelta, timezone
from uuid import UUID

import structlog
from sqlalchemy import or_, select, update

from src.config import get_settings
from src.core.database import async_session_factory
from src.domain.storage_ingest.archive_paths import staging_file_is_visible
from src.domain.storage_ingest.models import StorageIngestJob, StorageIngestJobStatus
from src.domain.storage_ingest.repository import StorageIngestJobRepository
from src.workers.tasks.storage_ingest import process_storage_ingest_job

logger = structlog.get_logger()

_RETRYABLE_FAIL_MARKERS = ("staging_file_missing", "worker_enqueue_failed")

_scheduled: set[str] = set()
_recovery_task: asyncio.Task | None = None


def schedule_storage_ingest_fallback(job_id: str, *, delay_seconds: int | None = None) -> None:
    """Process the job on this API worker if it is still waiting and the file is here."""
    if job_id in _scheduled:
        return
    _scheduled.add(job_id)
    asyncio.create_task(_watchdog(job_id, delay_seconds=delay_seconds))


def start_storage_ingest_background_tasks() -> None:
    """Start the periodic pending-job recovery loop once per API process."""
    global _recovery_task
    if _recovery_task and not _recovery_task.done():
        return
    _recovery_task = asyncio.create_task(_recovery_loop())


async def stop_storage_ingest_background_tasks() -> None:
    """Cancel the periodic recovery loop during app shutdown."""
    global _recovery_task
    task = _recovery_task
    _recovery_task = None
    if not task:
        return
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        pass


def _is_retryable_failure(error: str | None) -> bool:
    text = (error or "").lower()
    return any(marker in text for marker in _RETRYABLE_FAIL_MARKERS)


async def _reset_stale_running_jobs() -> int:
    """Return crashed in-flight jobs to pending so a host with the file can retry."""
    settings = get_settings()
    cutoff = datetime.now(timezone.utc) - timedelta(
        seconds=settings.storage_ingest_running_stale_seconds
    )
    try:
        async with async_session_factory() as db:
            result = await db.execute(
                update(StorageIngestJob)
                .where(
                    StorageIngestJob.status == StorageIngestJobStatus.RUNNING.value,
                    StorageIngestJob.updated_at < cutoff,
                )
                .values(
                    status=StorageIngestJobStatus.PENDING.value,
                    error=None,
                )
                .returning(StorageIngestJob.id)
            )
            ids = [str(row[0]) for row in result.all()]
            await db.commit()
    except Exception as exc:
        logger.error("storage_ingest_running_reset_failed", error=str(exc), exc_info=True)
        return 0

    for job_id in ids:
        logger.warning("storage_ingest_stale_running_reset", job_id=job_id)
        schedule_storage_ingest_fallback(job_id, delay_seconds=1)
    return len(ids)


async def _schedule_stale_pending_jobs() -> int:
    """Find waiting or retryable-failed jobs and schedule inline recovery."""
    settings = get_settings()
    grace = settings.storage_ingest_worker_grace_seconds
    cutoff = datetime.now(timezone.utc) - timedelta(seconds=grace)
    try:
        async with async_session_factory() as db:
            result = await db.execute(
                select(StorageIngestJob).where(
                    or_(
                        StorageIngestJob.status == StorageIngestJobStatus.PENDING.value,
                        StorageIngestJob.status == StorageIngestJobStatus.FAILED.value,
                    ),
                    StorageIngestJob.updated_at < cutoff,
                ).limit(100)
            )
            rows = list(result.scalars().all())
    except Exception as exc:
        logger.error("storage_ingest_resume_scan_failed", error=str(exc), exc_info=True)
        return 0

    scheduled = 0
    for row in rows:
        if row.status == StorageIngestJobStatus.FAILED.value and not _is_retryable_failure(row.error):
            continue
        logger.info("storage_ingest_resume_fallback", job_id=str(row.id), status=row.status)
        schedule_storage_ingest_fallback(str(row.id), delay_seconds=1)
        scheduled += 1
    return scheduled


async def _recovery_loop() -> None:
    """Continuously reschedule stale ingest jobs for inline processing."""
    await asyncio.sleep(10)
    while True:
        await _reset_stale_running_jobs()
        await _schedule_stale_pending_jobs()
        delay = max(10, get_settings().storage_ingest_worker_grace_seconds)
        await asyncio.sleep(delay)


async def _watchdog(job_id: str, *, delay_seconds: int | None = None) -> None:
    settings = get_settings()
    delay = settings.storage_ingest_worker_grace_seconds if delay_seconds is None else delay_seconds
    try:
        if delay > 0:
            await asyncio.sleep(delay)

        async with async_session_factory() as db:
            row = await StorageIngestJobRepository(db).get(UUID(job_id))
            if not row:
                return
            if row.status == StorageIngestJobStatus.COMPLETED.value:
                return
            if row.status == StorageIngestJobStatus.RUNNING.value:
                return
            if row.status == StorageIngestJobStatus.FAILED.value and not _is_retryable_failure(row.error):
                return
            if row.status not in (
                StorageIngestJobStatus.PENDING.value,
                StorageIngestJobStatus.FAILED.value,
            ):
                return
            if not staging_file_is_visible(settings.storage_staging_upload_root, row.staging_relative_path):
                logger.info(
                    "storage_ingest_fallback_file_not_on_this_host",
                    job_id=job_id,
                )
                return

        logger.warning(
            "storage_ingest_api_fallback_starting",
            job_id=job_id,
            reason="api_host_has_staging_file",
        )
        ctx = {"db_factory": async_session_factory}
        await process_storage_ingest_job(ctx, job_id)
    except Exception:
        logger.exception("storage_ingest_api_fallback_failed", job_id=job_id)
    finally:
        _scheduled.discard(job_id)