"""Durable asynchronous job queue with pause, resume, cancel and recovery."""

from __future__ import annotations

import asyncio
import logging
import time
from collections.abc import Awaitable, Callable
from dataclasses import dataclass

from app.config import Settings
from app.database.models import (
    Job,
    JobStatus,
    Video,
    VideoStatus,
    utc_now,
)
from app.database.repository import Repository
from app.errors import (
    AppError,
    DestinationError,
    DownloadCancelledError,
    DuplicateJobError,
    JobCancelledError,
    JobStateError,
    MediaNotFoundError,
    sanitize_error,
)
from app.services.crawler import Crawler
from app.services.downloader import Downloader
from app.services.uploader import Destination, Uploader
from app.utils.formatting import format_progress_text

logger = logging.getLogger(__name__)

Notifier = Callable[[int, str], Awaitable[None]]
_STOP = "__STOP__"


@dataclass(slots=True)
class JobControl:
    cancel_event: asyncio.Event
    resume_event: asyncio.Event
    paused: bool = False


class JobManager:
    def __init__(
        self,
        settings: Settings,
        repo: Repository,
        crawler: Crawler,
        downloader: Downloader,
        uploader: Uploader,
        *,
        notifier: Notifier | None = None,
    ) -> None:
        self.settings = settings
        self.repo = repo
        self.crawler = crawler
        self.downloader = downloader
        self.uploader = uploader
        self.notifier = notifier
        self._queue: asyncio.Queue[str] = asyncio.Queue()
        self._queued: set[str] = set()
        self._workers: list[asyncio.Task[None]] = []
        self._controls: dict[str, JobControl] = {}
        self._last_notify: dict[str, float] = {}
        self._notify_lock = asyncio.Lock()
        self._started = False
        self._stopping = False

    # ------------------------------------------------------------- lifecycle
    async def start(self) -> None:
        if self._started:
            return
        self._stopping = False
        pending = await self.repo.recover_after_restart()
        self._started = True
        for index in range(self.settings.max_concurrent_downloads):
            task = asyncio.create_task(self._worker(index), name=f"job-worker-{index}")
            self._workers.append(task)
        for job_id in pending:
            await self.enqueue(job_id)
        logger.info(
            "job manager started with %s worker(s); %s job(s) queued",
            len(self._workers),
            len(pending),
        )

    async def stop(self, grace_seconds: float = 45.0) -> None:
        if not self._started:
            return
        self._stopping = True
        for control in list(self._controls.values()):
            control.cancel_event.set()
        # Drain queued jobs; they stay PENDING in SQLite for the next startup.
        while True:
            try:
                item = self._queue.get_nowait()
            except asyncio.QueueEmpty:
                break
            self._queue.task_done()
            self._queued.discard(item)
        for _ in self._workers:
            await self._queue.put(_STOP)
        if self._workers:
            try:
                await asyncio.wait_for(
                    asyncio.gather(*self._workers, return_exceptions=True), timeout=grace_seconds
                )
            except TimeoutError:
                logger.warning("job workers did not stop in time; cancelling tasks")
                for task in self._workers:
                    task.cancel()
                await asyncio.gather(*self._workers, return_exceptions=True)
        self._workers.clear()
        self._started = False
        logger.info("job manager stopped")

    async def wait_until_idle(self, wait_seconds: float = 30.0) -> bool:
        deadline = time.monotonic() + wait_seconds
        while time.monotonic() < deadline:
            if not self._controls and self._queue.empty():
                return True
            await asyncio.sleep(0.05)
        return False

    # ----------------------------------------------------------- public API
    async def create_job(self, source_url: str, owner_id: int, owner_chat_id: int) -> Job:
        validated = await self.crawler.guard.validate(source_url, resolve_dns=True)
        source_url = validated.url
        existing = await self.repo.find_active_job_for_source(source_url)
        if existing is not None:
            raise DuplicateJobError(
                user_message=f"یک عملیات فعال برای همین آدرس وجود دارد: {existing.short_id}"
            )
        job = Job.create(source_url, owner_id=owner_id, owner_chat_id=owner_chat_id)
        await self.repo.create_job(job)
        await self.enqueue(job.id)
        await self.repo.refresh_job_counts(job.id)
        fresh = await self.repo.get_job(job.id) or job
        await self._notify(fresh, force=True)
        logger.info(
            "job created",
            extra={"job_id": job.id[:8]},
        )
        return fresh

    async def enqueue(self, job_id: str) -> None:
        if not self._started or self._stopping:
            return
        if job_id in self._queued:
            return
        job = await self.repo.get_job(job_id)
        if job is None or job.status != JobStatus.PENDING:
            return
        self._queued.add(job_id)
        await self._queue.put(job_id)

    async def get_job(self, job_id: str) -> Job | None:
        return await self.repo.get_job(job_id)

    async def active_jobs(self, owner_id: int | None = None) -> list[Job]:
        return await self.repo.get_active_jobs(owner_id)

    async def select_job(self, job_id: str | None, owner_id: int | None = None) -> Job:
        if job_id:
            job = await self.repo.get_job(job_id)
            if job is None:
                raise JobStateError("عملیات با این شناسه پیدا نشد.")
            if owner_id is not None and job.owner_id != owner_id:
                raise JobStateError("این عملیات متعلق به شما نیست.")
            return job
        job = await self.repo.latest_active_job(owner_id)
        if job is None:
            raise JobStateError("هیچ عملیات فعالی وجود ندارد.")
        return job

    async def pause_job(self, job_id: str | None = None, owner_id: int | None = None) -> Job:
        job = await self.select_job(job_id, owner_id)
        if job.status in {JobStatus.COMPLETED, JobStatus.FAILED, JobStatus.CANCELLED}:
            raise JobStateError("عملیات پایان‌یافته قابل توقف نیست.")
        if job.status == JobStatus.PAUSED:
            return job
        control = self._controls.get(job.id)
        if control is not None:
            control.paused = True
            control.resume_event.clear()
        await self.repo.update_job_fields(
            job.id, status=JobStatus.PAUSED, error_message=None
        )
        fresh = await self.repo.get_job(job.id) or job
        await self._notify(fresh, force=True)
        logger.info("job paused", extra={"job_id": job.id[:8]})
        return fresh

    async def resume_job(self, job_id: str | None = None, owner_id: int | None = None) -> Job:
        if job_id is None:
            paused = await self.repo.list_jobs(limit=1, statuses=[JobStatus.PAUSED])
            job = paused[0] if paused else None
            if job is None:
                raise JobStateError("هیچ عملیات متوقف‌شده‌ای برای ادامه وجود ندارد.")
        else:
            job = await self.select_job(job_id, owner_id)
        if job.status != JobStatus.PAUSED:
            raise JobStateError("فقط عملیات متوقف‌شده قابل ادامه است.")
        control = self._controls.get(job.id)
        if control is not None:
            control.paused = False
            control.resume_event.set()
            await self.repo.update_job_fields(
                job.id, status=JobStatus.RUNNING, error_message=None
            )
        else:
            await self.repo.update_job_fields(
                job.id, status=JobStatus.PENDING, error_message=None
            )
            await self.enqueue(job.id)
        fresh = await self.repo.get_job(job.id) or job
        await self._notify(fresh, force=True)
        logger.info("job resumed", extra={"job_id": job.id[:8]})
        return fresh

    async def cancel_job(self, job_id: str | None = None, owner_id: int | None = None) -> Job:
        job = await self.select_job(job_id, owner_id)
        if job.status in {JobStatus.COMPLETED, JobStatus.FAILED, JobStatus.CANCELLED}:
            raise JobStateError("این عملیات از قبل پایان یافته است.")
        control = self._controls.get(job.id)
        if control is not None:
            control.cancel_event.set()
        await self.repo.set_job_status(
            job.id,
            JobStatus.CANCELLED,
            error_message="توسط مدیر لغو شد.",
            finished=True,
        )
        await self.repo.update_job_fields(job.id, current_stage="done")
        await self.repo.cancel_pending_videos(job.id)
        await self.repo.refresh_job_counts(job.id)
        fresh = await self.repo.get_job(job.id) or job
        await self._notify(fresh, force=True)
        logger.info("job cancelled", extra={"job_id": job.id[:8]})
        return fresh

    # --------------------------------------------------------------- workers
    async def _worker(self, index: int) -> None:
        logger.debug("worker %s started", index)
        while True:
            item = await self._queue.get()
            try:
                if item == _STOP:
                    return
                self._queued.discard(item)
                try:
                    await self._run_job(item)
                except asyncio.CancelledError:
                    raise
                except Exception:
                    logger.exception(
                        "worker failed unexpectedly", extra={"job_id": item[:8]}
                    )
            finally:
                self._queue.task_done()
                logger.debug("worker %s waiting", index)

    async def _run_job(self, job_id: str) -> None:
        job = await self.repo.get_job(job_id)
        if job is None:
            return
        if job.status != JobStatus.PENDING:
            logger.debug("skipping non-pending job", extra={"job_id": job_id[:8]})
            return
        if self._stopping:
            return
        control = JobControl(cancel_event=asyncio.Event(), resume_event=asyncio.Event())
        control.resume_event.set()
        self._controls[job_id] = control
        try:
            if not await self.repo.start_job(job_id):
                return
            job = await self.repo.get_job(job_id)
            if job is None:
                return
            logger.info("job started", extra={"job_id": job_id[:8]})
            await self._process_job(job, control)
        except (JobCancelledError, DownloadCancelledError) as exc:
            await self._handle_cancel(job_id, exc)
        except asyncio.CancelledError:
            if not self._stopping:
                logger.warning("job task cancelled", extra={"job_id": job_id[:8]})
                await self._fail_job(job_id, JobCancelledError("worker cancelled"))
            raise
        except Exception as exc:  # noqa: BLE001
            if self._stopping and control.cancel_event.is_set():
                await self._handle_cancel(job_id, exc)
            else:
                logger.exception("job failed", extra={"job_id": job_id[:8]})
                await self._fail_job(job_id, exc)
        finally:
            self._controls.pop(job_id, None)
            self._last_notify.pop(job_id, None)

    async def _process_job(self, job: Job, control: JobControl) -> None:
        started_monotonic = time.monotonic()
        deadline = started_monotonic + self.settings.max_job_seconds
        await self._checkpoint(job.id)
        try:
            destination = await self.uploader.resolve_destination()
        except DestinationError as exc:
            raise AppError(f"destination invalid: {exc}") from exc

        await self.repo.update_job_fields(job.id, error_message=None)
        await self._notify(job, force=False)

        if job.current_stage in {"queued", "crawl"}:
            await self.repo.set_job_stage(job.id, "crawl")
            result = await self.crawler.crawl(
                job, control.cancel_event, lambda: self._checkpoint(job.id)
            )
            logger.info(
                "crawl stage complete: %s videos, stop=%s",
                result.videos_found,
                result.stop_reason,
                extra={"job_id": job.id[:8]},
            )
            await self.repo.set_job_stage(job.id, "media")
            await self.repo.refresh_job_counts(job.id)
            await self._notify(job, force=False)

        videos = await self.repo.list_videos(job.id)
        if not videos:
            raise MediaNotFoundError("no videos were stored for this job")
        await self.repo.set_job_stage(job.id, "download_upload")

        fatal_error: Exception | None = None
        for snapshot in videos:
            await self._checkpoint(job.id)
            if time.monotonic() > deadline:
                fatal_error = TimeoutError("max job duration reached")
                break
            current = await self.repo.get_video(snapshot.id)
            if current is None or current.status in {
                VideoStatus.UPLOADED,
                VideoStatus.SKIPPED,
                VideoStatus.CANCELLED,
                VideoStatus.UPLOAD_UNKNOWN,
                VideoStatus.FAILED,
            }:
                continue
            if current.status == VideoStatus.PENDING:
                fatal = await self._download_one(current, control.cancel_event)
                if fatal is not None:
                    fatal_error = fatal
                    break
                current = await self.repo.get_video(snapshot.id)
                if current is None or current.status != VideoStatus.DOWNLOADED:
                    await self.repo.refresh_job_counts(job.id)
                    await self._notify(job, force=False)
                    continue
            if current.status == VideoStatus.DOWNLOADED:
                await self._checkpoint(job.id)
                fatal = await self._upload_one(current, destination, control.cancel_event)
                if fatal is not None:
                    fatal_error = fatal
                    break
            await self.repo.refresh_job_counts(job.id)
            await self._notify(job, force=False)

        await self._finish_job(job.id, fatal_error)

    async def _download_one(
        self, video: Video, cancel_event: asyncio.Event
    ) -> Exception | None:
        claimed = await self.repo.claim_video(
            video.id, (VideoStatus.PENDING,), VideoStatus.DOWNLOADING
        )
        if not claimed:
            return None
        fresh = await self.repo.get_video(video.id) or video
        try:
            result = await self.downloader.prepare(fresh, cancel_event)
            await self.repo.update_video_fields(
                video.id,
                status=VideoStatus.DOWNLOADED,
                local_path=None,
                file_size=result.size if result.size is not None else fresh.file_size,
                media_type=result.mime_type,
                duration_seconds=result.duration_seconds or fresh.duration_seconds,
                error_message=None,
            )
            logger.info(
                "video source validated for direct streaming (size=%s)",
                result.size,
                extra={"job_id": video.job_id[:8]},
            )
            return None
        except (DownloadCancelledError, JobCancelledError):
            raise
        except AppError as exc:
            await self.repo.update_video_fields(
                video.id,
                status=VideoStatus.FAILED,
                error_message=sanitize_error(exc),
            )
            await self.repo.refresh_job_counts(video.job_id)
            logger.warning(
                "video preparation failed: %s",
                sanitize_error(exc),
                extra={"job_id": video.job_id[:8]},
            )
            return None
        except Exception as exc:  # noqa: BLE001
            await self.repo.update_video_fields(
                video.id,
                status=VideoStatus.FAILED,
                error_message=sanitize_error(exc),
            )
            await self.repo.refresh_job_counts(video.job_id)
            logger.exception(
                "unexpected video preparation failure", extra={"job_id": video.job_id[:8]}
            )
            return None

    async def _upload_one(
        self,
        video: Video,
        destination: Destination,
        cancel_event: asyncio.Event,
    ) -> Exception | None:
        claimed = await self.repo.claim_video(
            video.id, (VideoStatus.DOWNLOADED,), VideoStatus.UPLOADING
        )
        if not claimed:
            return None
        try:
            result = await self.uploader.upload(video, destination, cancel_event)
            await self.repo.update_video_fields(
                video.id,
                status=VideoStatus.UPLOADED,
                telegram_message_id=result.message_id,
                uploaded_at=utc_now(),
                error_message=None,
            )
            logger.info(
                "video uploaded: message_id=%s",
                result.message_id,
                extra={"job_id": video.job_id[:8]},
            )
            return None
        except (JobCancelledError, DownloadCancelledError):
            raise
        except AppError as exc:
            uncertain = exc.code == "upload_uncertain"
            status = VideoStatus.UPLOAD_UNKNOWN if uncertain else VideoStatus.FAILED
            await self.repo.update_video_fields(
                video.id,
                status=status,
                error_message=sanitize_error(exc),
            )
            await self.repo.refresh_job_counts(video.job_id)
            logger.warning(
                "video upload ended as %s: %s",
                status.value,
                sanitize_error(exc),
                extra={"job_id": video.job_id[:8]},
            )
            return None
        except Exception as exc:  # noqa: BLE001
            await self.repo.update_video_fields(
                video.id,
                status=VideoStatus.UPLOAD_UNKNOWN,
                error_message=sanitize_error(exc),
            )
            logger.exception("unexpected upload failure", extra={"job_id": video.job_id[:8]})
            return None

    async def _finish_job(self, job_id: str, fatal_error: Exception | None) -> None:
        job = await self.repo.get_job(job_id)
        if job is None:
            return
        if job.status == JobStatus.CANCELLED:
            return
        await self.repo.refresh_job_counts(job_id)
        job = await self.repo.get_job(job_id)
        if job is None:
            return
        if fatal_error is not None:
            await self.repo.cancel_unfinished_videos(job_id)
            await self.repo.refresh_job_counts(job_id)
            await self.repo.update_job_fields(
                job_id,
                status=JobStatus.FAILED,
                finished_at=utc_now(),
                current_stage="done",
                error_message=sanitize_error(fatal_error),
            )
        elif job.uploaded_videos > 0:
            error_message = None
            if job.failed_videos:
                error_message = (
                    f"{job.failed_videos} ویدئو ناموفق یا با وضعیت ارسال نامشخص بود."
                )
            await self.repo.update_job_fields(
                job_id,
                status=JobStatus.COMPLETED,
                finished_at=utc_now(),
                current_stage="done",
                error_message=error_message,
            )
        else:
            await self.repo.cancel_unfinished_videos(job_id)
            await self.repo.update_job_fields(
                job_id,
                status=JobStatus.FAILED,
                finished_at=utc_now(),
                current_stage="done",
                error_message="هیچ ویدئویی با موفقیت ارسال نشد.",
            )
        await self.repo.refresh_job_counts(job_id)
        fresh = await self.repo.get_job(job_id)
        if fresh is not None:
            logger.info(
                "job finished with status %s", fresh.status.value, extra={"job_id": job_id[:8]}
            )
            await self._notify(fresh, force=True)

    async def _fail_job(self, job_id: str, exc: BaseException) -> None:
        job = await self.repo.get_job(job_id)
        if job is None or job.status == JobStatus.CANCELLED:
            return
        await self.repo.cancel_unfinished_videos(job_id)
        await self.repo.refresh_job_counts(job_id)
        await self.repo.update_job_fields(
            job_id,
            status=JobStatus.FAILED,
            finished_at=utc_now(),
            current_stage="done",
            error_message=sanitize_error(exc),
        )
        await self.repo.refresh_job_counts(job_id)
        fresh = await self.repo.get_job(job_id)
        if fresh is not None:
            await self._notify(fresh, force=True)

    async def _handle_cancel(self, job_id: str, exc: BaseException) -> None:
        job = await self.repo.get_job(job_id)
        if job is None:
            return
        if self._stopping:
            if job.status == JobStatus.RUNNING:
                await self.repo.update_job_fields(
                    job_id,
                    status=JobStatus.PENDING,
                    error_message="برنامه به‌صورت امن متوقف شد؛ Job پس از اجرای بعدی ادامه می‌یابد.",
                )
            await self.repo.reset_interrupted_videos(job_id)
            await self.repo.refresh_job_counts(job_id)
            logger.info("job preserved for restart", extra={"job_id": job_id[:8]})
            return
        if job.status != JobStatus.CANCELLED:
            await self.repo.set_job_status(
                job_id,
                JobStatus.CANCELLED,
                error_message="توسط مدیر لغو شد.",
                finished=True,
            )
        await self.repo.update_job_fields(job_id, current_stage="done")
        await self.repo.cancel_unfinished_videos(job_id)
        await self.repo.refresh_job_counts(job_id)
        fresh = await self.repo.get_job(job_id)
        if fresh is not None:
            logger.info("job stopped by cancellation", extra={"job_id": job_id[:8]})
            await self._notify(fresh, force=True)

    async def _checkpoint(self, job_id: str) -> None:
        control = self._controls.get(job_id)
        if control is None:
            return
        if control.cancel_event.is_set():
            raise JobCancelledError()
        if not control.paused:
            return
        job = await self.repo.get_job(job_id)
        if job is not None and job.status == JobStatus.CANCELLED:
            raise JobCancelledError()
        if job is not None and job.status == JobStatus.RUNNING:
            await self.repo.update_job_fields(
                job_id, status=JobStatus.PAUSED, error_message=None
            )
        while control.paused:
            if control.cancel_event.is_set():
                raise JobCancelledError()
            try:
                await asyncio.wait_for(control.resume_event.wait(), timeout=0.5)
            except TimeoutError:
                continue
        if control.cancel_event.is_set():
            raise JobCancelledError()
        latest = await self.repo.get_job(job_id)
        if latest is None or latest.status == JobStatus.CANCELLED:
            raise JobCancelledError()
        if latest.status == JobStatus.PAUSED:
            await self.repo.update_job_fields(job_id, status=JobStatus.RUNNING)

    async def _notify(self, job: Job, *, force: bool = False) -> None:
        if self.notifier is None:
            return
        now = time.monotonic()
        async with self._notify_lock:
            last = self._last_notify.get(job.id, 0.0)
            if not force and now - last < self.settings.progress_update_interval:
                return
            self._last_notify[job.id] = now
            fresh = await self.repo.get_job(job.id) or job
            text = format_progress_text(fresh)
            try:
                await self.notifier(fresh.owner_chat_id, text)
            except Exception:
                logger.warning("could not send progress notification", extra={"job_id": job.id[:8]})