"""Transactional repositories for jobs, videos, pages and runtime settings."""

from __future__ import annotations

import logging
from collections.abc import Iterable
from typing import Any, TypeVar

import aiosqlite

from app.database.database import Database
from app.database.models import (
    ACTIVE_JOB_STATUSES,
    TERMINAL_VIDEO_STATUSES,
    Job,
    JobStatus,
    Page,
    PageStatus,
    Video,
    VideoStatus,
    utc_now,
)

logger = logging.getLogger(__name__)
T = TypeVar("T")

_JOB_FIELDS = {
    "status",
    "started_at",
    "finished_at",
    "total_pages",
    "processed_pages",
    "total_videos",
    "processed_videos",
    "downloaded_videos",
    "uploaded_videos",
    "failed_videos",
    "current_stage",
    "error_message",
}
_VIDEO_FIELDS = {
    "job_id",
    "status",
    "title",
    "caption",
    "media_type",
    "file_size",
    "media_url",
    "thumbnail_url",
    "duration_seconds",
    "local_path",
    "telegram_message_id",
    "uploaded_at",
    "error_message",
    "attempts",
    "updated_at",
}


def _value(value: Any) -> Any:
    if isinstance(value, (JobStatus, VideoStatus, PageStatus)):
        return value.value
    return value


def _job_from_row(row: aiosqlite.Row) -> Job:
    return Job(
        id=row["id"],
        source_url=row["source_url"],
        status=JobStatus(row["status"]),
        owner_id=row["owner_id"],
        owner_chat_id=row["owner_chat_id"],
        correlation_id=row["correlation_id"],
        created_at=row["created_at"],
        started_at=row["started_at"],
        finished_at=row["finished_at"],
        total_pages=row["total_pages"],
        processed_pages=row["processed_pages"],
        total_videos=row["total_videos"],
        processed_videos=row["processed_videos"],
        downloaded_videos=row["downloaded_videos"],
        uploaded_videos=row["uploaded_videos"],
        failed_videos=row["failed_videos"],
        current_stage=row["current_stage"],
        error_message=row["error_message"],
        updated_at=row["updated_at"],
    )


def _video_from_row(row: aiosqlite.Row) -> Video:
    return Video(
        id=row["id"],
        job_id=row["job_id"],
        source_url=row["source_url"],
        page_url=row["page_url"],
        title=row["title"],
        caption=row["caption"],
        media_type=row["media_type"],
        file_size=row["file_size"],
        status=VideoStatus(row["status"]),
        telegram_message_id=row["telegram_message_id"],
        created_at=row["created_at"],
        uploaded_at=row["uploaded_at"],
        updated_at=row["updated_at"],
        media_url=row["media_url"],
        thumbnail_url=row["thumbnail_url"],
        duration_seconds=row["duration_seconds"],
        local_path=row["local_path"],
        error_message=row["error_message"],
        attempts=row["attempts"],
    )


def _page_from_row(row: aiosqlite.Row) -> Page:
    return Page(
        id=row["id"],
        job_id=row["job_id"],
        page_url=row["page_url"],
        status=PageStatus(row["status"]),
        discovered_at=row["discovered_at"],
        processed_at=row["processed_at"],
        http_status=row["http_status"],
        error_message=row["error_message"],
    )


class Repository:
    def __init__(self, database: Database) -> None:
        self.db = database

    # ------------------------------------------------------------------ jobs
    async def create_job(self, job: Job) -> Job:
        async with self.db.transaction() as conn:
            await conn.execute(
                """
                INSERT INTO jobs (
                    id, source_url, status, owner_id, owner_chat_id, correlation_id,
                    created_at, updated_at, current_stage
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    job.id,
                    job.source_url,
                    job.status.value,
                    job.owner_id,
                    job.owner_chat_id,
                    job.correlation_id,
                    job.created_at,
                    job.updated_at,
                    job.current_stage,
                ),
            )
        return job

    async def get_job(self, job_id: str) -> Job | None:
        row = await self.db.fetch_one("SELECT * FROM jobs WHERE id = ?", (job_id,))
        return _job_from_row(row) if row else None

    async def find_active_job_for_source(self, source_url: str) -> Job | None:
        placeholders = ",".join("?" for _ in ACTIVE_JOB_STATUSES)
        row = await self.db.fetch_one(
            f"SELECT * FROM jobs WHERE source_url = ? AND status IN ({placeholders}) "
            "ORDER BY created_at DESC LIMIT 1",
            (source_url, *(s.value for s in ACTIVE_JOB_STATUSES)),
        )
        return _job_from_row(row) if row else None

    async def list_jobs(
        self,
        limit: int = 10,
        statuses: Iterable[JobStatus] | None = None,
    ) -> list[Job]:
        if statuses is not None:
            values = tuple(s.value for s in statuses)
            if not values:
                return []
            placeholders = ",".join("?" for _ in values)
            rows = await self.db.fetch_all(
                f"SELECT * FROM jobs WHERE status IN ({placeholders}) "
                "ORDER BY created_at DESC LIMIT ?",
                (*values, limit),
            )
        else:
            rows = await self.db.fetch_all(
                "SELECT * FROM jobs ORDER BY created_at DESC LIMIT ?", (limit,)
            )
        return [_job_from_row(row) for row in rows]

    async def get_active_jobs(self, owner_id: int | None = None) -> list[Job]:
        values = tuple(s.value for s in ACTIVE_JOB_STATUSES)
        placeholders = ",".join("?" for _ in values)
        if owner_id is None:
            rows = await self.db.fetch_all(
                f"SELECT * FROM jobs WHERE status IN ({placeholders}) "
                "ORDER BY created_at DESC",
                values,
            )
        else:
            rows = await self.db.fetch_all(
                f"SELECT * FROM jobs WHERE owner_id = ? AND status IN ({placeholders}) "
                "ORDER BY created_at DESC",
                (owner_id, *values),
            )
        return [_job_from_row(row) for row in rows]

    async def latest_active_job(self, owner_id: int | None = None) -> Job | None:
        jobs = await self.get_active_jobs(owner_id)
        return jobs[0] if jobs else None

    async def start_job(self, job_id: str) -> bool:
        """Atomically move PENDING -> RUNNING; prevents concurrent duplicate runs."""

        now = utc_now()
        async with self.db.transaction() as conn:
            cursor = await conn.execute(
                """
                UPDATE jobs SET status = 'RUNNING',
                    started_at = COALESCE(started_at, ?),
                    updated_at = ?,
                    current_stage = CASE WHEN current_stage = 'queued' THEN 'crawl'
                                         ELSE current_stage END
                WHERE id = ? AND status = 'PENDING'
                """,
                (now, now, job_id),
            )
            return cursor.rowcount == 1

    async def update_job_fields(self, job_id: str, **fields: Any) -> bool:
        updates = {key: _value(value) for key, value in fields.items() if key in _JOB_FIELDS}
        if not updates:
            return False
        updates["updated_at"] = utc_now()
        assignments = ", ".join(f"{key} = ?" for key in updates)
        async with self.db.transaction() as conn:
            cursor = await conn.execute(
                f"UPDATE jobs SET {assignments} WHERE id = ?",
                (*updates.values(), job_id),
            )
            return cursor.rowcount == 1

    async def set_job_status(
        self,
        job_id: str,
        status: JobStatus,
        *,
        error_message: str | None = None,
        finished: bool = False,
    ) -> None:
        fields: dict[str, Any] = {"status": status, "error_message": error_message}
        if finished:
            fields["finished_at"] = utc_now()
        await self.update_job_fields(job_id, **fields)

    async def set_job_stage(self, job_id: str, stage: str) -> None:
        await self.update_job_fields(job_id, current_stage=stage)

    async def refresh_job_counts(self, job_id: str) -> Job | None:
        terminal = tuple(s.value for s in TERMINAL_VIDEO_STATUSES)
        terminal_ph = ",".join("?" for _ in terminal)
        downloaded = (VideoStatus.DOWNLOADED.value, VideoStatus.UPLOADED.value,
                      VideoStatus.UPLOADING.value, VideoStatus.UPLOAD_UNKNOWN.value)
        downloaded_ph = ",".join("?" for _ in downloaded)
        now = utc_now()
        async with self.db.transaction() as conn:
            await conn.execute(
                f"""
                UPDATE jobs SET
                    total_pages = (SELECT COUNT(*) FROM pages WHERE job_id = ?),
                    processed_pages = (SELECT COUNT(*) FROM pages WHERE job_id = ?
                        AND status IN ('PROCESSED','SKIPPED')),
                    total_videos = (SELECT COUNT(*) FROM videos WHERE job_id = ?),
                    processed_videos = (SELECT COUNT(*) FROM videos WHERE job_id = ?
                        AND status IN ({terminal_ph})),
                    downloaded_videos = (SELECT COUNT(*) FROM videos WHERE job_id = ?
                        AND status IN ({downloaded_ph})),
                    uploaded_videos = (SELECT COUNT(*) FROM videos WHERE job_id = ?
                        AND status = 'UPLOADED'),
                    failed_videos = (SELECT COUNT(*) FROM videos WHERE job_id = ?
                        AND status IN ('FAILED','UPLOAD_UNKNOWN')),
                    updated_at = ?
                WHERE id = ?
                """,
                (
                    job_id,
                    job_id,
                    job_id,
                    job_id,
                    *terminal,
                    job_id,
                    *downloaded,
                    job_id,
                    job_id,
                    now,
                    job_id,
                ),
            )
        return await self.get_job(job_id)

    async def recover_after_restart(self) -> list[str]:
        """Make crash-interrupted work resumable, conservatively."""

        now = utc_now()
        async with self.db.transaction() as conn:
            await conn.execute(
                "UPDATE videos SET status = 'PENDING', error_message = NULL, updated_at = ? "
                "WHERE status = 'DOWNLOADING'",
                (now,),
            )
            await conn.execute(
                "UPDATE videos SET status = 'UPLOAD_UNKNOWN', updated_at = ? "
                "WHERE status = 'UPLOADING'",
                (now,),
            )
            await conn.execute(
                "UPDATE pages SET status = 'PENDING', error_message = NULL "
                "WHERE status = 'PROCESSING'"
            )
            await conn.execute(
                """
                UPDATE jobs SET
                    status = 'PENDING',
                    current_stage = CASE WHEN current_stage = 'queued' THEN 'crawl'
                                         ELSE current_stage END,
                    error_message = 'پس از راه‌اندازی مجدد بازیابی شد.',
                    updated_at = ?
                WHERE status = 'RUNNING'
                """,
                (now,),
            )
            cursor = await conn.execute(
                "SELECT id FROM jobs WHERE status = 'PENDING' ORDER BY created_at ASC"
            )
            rows = await cursor.fetchall()
            await cursor.close()
        recovered = [row["id"] for row in rows]
        if recovered:
            logger.info("recovered %s pending job(s) after restart", len(recovered))
        return recovered

    # ----------------------------------------------------------------- pages
    async def add_page(self, job_id: str, page_url: str) -> tuple[Page, bool]:
        page = Page.create(job_id, page_url)
        async with self.db.transaction() as conn:
            cursor = await conn.execute(
                """
                INSERT OR IGNORE INTO pages
                    (id, job_id, page_url, status, discovered_at)
                VALUES (?, ?, ?, ?, ?)
                """,
                (page.id, job_id, page_url, page.status.value, page.discovered_at),
            )
            created = cursor.rowcount == 1
            if not created:
                cursor = await conn.execute(
                    "SELECT * FROM pages WHERE job_id = ? AND page_url = ?",
                    (job_id, page_url),
                )
                row = await cursor.fetchone()
                await cursor.close()
                if row is None:  # pragma: no cover - defensive
                    raise RuntimeError("page insert conflicted but row was not found")
                page = _page_from_row(row)
        return page, created

    async def update_page(
        self,
        page_id: str,
        status: PageStatus,
        *,
        http_status: int | None = None,
        error_message: str | None = None,
    ) -> None:
        processed_at = utc_now() if status in {PageStatus.PROCESSED, PageStatus.SKIPPED} else None
        async with self.db.transaction() as conn:
            await conn.execute(
                """
                UPDATE pages SET status = ?, http_status = COALESCE(?, http_status),
                    error_message = ?, processed_at = COALESCE(?, processed_at)
                WHERE id = ?
                """,
                (status.value, http_status, error_message, processed_at, page_id),
            )

    async def list_pages(self, job_id: str) -> list[Page]:
        rows = await self.db.fetch_all(
            "SELECT * FROM pages WHERE job_id = ? ORDER BY discovered_at ASC", (job_id,)
        )
        return [_page_from_row(row) for row in rows]

    # ---------------------------------------------------------------- videos
    async def add_video(self, video: Video) -> tuple[Video, bool]:
        """Insert a globally unique video identity without ever duplicating reposts."""

        async with self.db.transaction() as conn:
            cursor = await conn.execute(
                """
                INSERT OR IGNORE INTO videos (
                    id, job_id, source_url, page_url, title, caption, media_type,
                    file_size, status, created_at, updated_at, media_url,
                    thumbnail_url, duration_seconds
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    video.id,
                    video.job_id,
                    video.source_url,
                    video.page_url,
                    video.title,
                    video.caption,
                    video.media_type,
                    video.file_size,
                    video.status.value,
                    video.created_at,
                    video.updated_at,
                    video.media_url,
                    video.thumbnail_url,
                    video.duration_seconds,
                ),
            )
            created = cursor.rowcount == 1
            if not created:
                cursor = await conn.execute(
                    "SELECT * FROM videos WHERE source_url = ?", (video.source_url,)
                )
                row = await cursor.fetchone()
                await cursor.close()
                if row is None:
                    raise RuntimeError("video insert conflicted but row was not found")
                video = _video_from_row(row)
        return video, created

    async def get_video(self, video_id: str) -> Video | None:
        row = await self.db.fetch_one("SELECT * FROM videos WHERE id = ?", (video_id,))
        return _video_from_row(row) if row else None

    async def get_video_by_source(self, source_url: str) -> Video | None:
        row = await self.db.fetch_one(
            "SELECT * FROM videos WHERE source_url = ?", (source_url,)
        )
        return _video_from_row(row) if row else None

    async def list_videos(
        self,
        job_id: str,
        statuses: Iterable[VideoStatus] | None = None,
    ) -> list[Video]:
        if statuses is not None:
            values = tuple(s.value for s in statuses)
            if not values:
                return []
            placeholders = ",".join("?" for _ in values)
            rows = await self.db.fetch_all(
                f"SELECT * FROM videos WHERE job_id = ? AND status IN ({placeholders}) "
                "ORDER BY created_at ASC",
                (job_id, *values),
            )
        else:
            rows = await self.db.fetch_all(
                "SELECT * FROM videos WHERE job_id = ? ORDER BY created_at ASC", (job_id,)
            )
        return [_video_from_row(row) for row in rows]

    async def update_video_fields(self, video_id: str, **fields: Any) -> bool:
        updates = {key: _value(value) for key, value in fields.items() if key in _VIDEO_FIELDS}
        if not updates:
            return False
        updates.setdefault("updated_at", utc_now())
        assignments = ", ".join(f"{key} = ?" for key in updates)
        async with self.db.transaction() as conn:
            cursor = await conn.execute(
                f"UPDATE videos SET {assignments} WHERE id = ?",
                (*updates.values(), video_id),
            )
            return cursor.rowcount == 1

    async def claim_video(
        self,
        video_id: str,
        from_statuses: Iterable[VideoStatus],
        to_status: VideoStatus,
    ) -> bool:
        values = tuple(s.value for s in from_statuses)
        if not values:
            return False
        placeholders = ",".join("?" for _ in values)
        now = utc_now()
        async with self.db.transaction() as conn:
            cursor = await conn.execute(
                f"""
                UPDATE videos SET status = ?, updated_at = ?,
                    attempts = attempts + CASE WHEN ? = 'DOWNLOADING' THEN 1 ELSE 0 END
                WHERE id = ? AND status IN ({placeholders})
                """,
                (to_status.value, now, to_status.value, video_id, *values),
            )
            return cursor.rowcount == 1

    async def cancel_pending_videos(self, job_id: str) -> None:
        async with self.db.transaction() as conn:
            await conn.execute(
                "UPDATE videos SET status = 'CANCELLED', updated_at = ? "
                "WHERE job_id = ? AND status IN ('PENDING','DOWNLOADED')",
                (utc_now(), job_id),
            )

    async def cancel_unfinished_videos(self, job_id: str) -> None:
        async with self.db.transaction() as conn:
            await conn.execute(
                "UPDATE videos SET status = 'CANCELLED', updated_at = ? "
                "WHERE job_id = ? AND status IN "
                "('PENDING','DOWNLOADING','DOWNLOADED','UPLOADING')",
                (utc_now(), job_id),
            )

    async def reset_interrupted_videos(self, job_id: str) -> None:
        """After a graceful stop, make downloads resumable and uploads conservative."""

        now = utc_now()
        async with self.db.transaction() as conn:
            await conn.execute(
                "UPDATE videos SET status = 'PENDING', updated_at = ? "
                "WHERE job_id = ? AND status = 'DOWNLOADING'",
                (now, job_id),
            )
            await conn.execute(
                "UPDATE videos SET status = 'UPLOAD_UNKNOWN', updated_at = ? "
                "WHERE job_id = ? AND status = 'UPLOADING'",
                (now, job_id),
            )

    # --------------------------------------------------------------- settings
    async def get_setting(self, key: str, default: str | None = None) -> str | None:
        row = await self.db.fetch_one("SELECT value FROM settings WHERE key = ?", (key,))
        return row["value"] if row else default

    async def set_setting(self, key: str, value: str) -> None:
        async with self.db.transaction() as conn:
            await conn.execute(
                """
                INSERT INTO settings(key, value, updated_at) VALUES (?, ?, ?)
                ON CONFLICT(key) DO UPDATE SET value = excluded.value,
                    updated_at = excluded.updated_at
                """,
                (key, value, utc_now()),
            )

    async def delete_setting(self, key: str) -> None:
        async with self.db.transaction() as conn:
            await conn.execute("DELETE FROM settings WHERE key = ?", (key,))

    # ---------------------------------------------------------------- reports
    async def job_status_counts(self) -> dict[str, int]:
        rows = await self.db.fetch_all("SELECT status, COUNT(*) AS n FROM jobs GROUP BY status")
        return {row["status"]: int(row["n"]) for row in rows}

    async def total_uploaded_videos(self) -> int:
        row = await self.db.fetch_one(
            "SELECT COUNT(*) AS n FROM videos WHERE status = 'UPLOADED'"
        )
        return int(row["n"]) if row else 0

    async def video_status_counts(self, job_id: str) -> dict[str, int]:
        rows = await self.db.fetch_all(
            "SELECT status, COUNT(*) AS n FROM videos WHERE job_id = ? GROUP BY status",
            (job_id,),
        )
        return {row["status"]: int(row["n"]) for row in rows}

    async def recent_uploaded(self, limit: int = 10) -> list[Video]:
        rows = await self.db.fetch_all(
            "SELECT * FROM videos WHERE status = 'UPLOADED' ORDER BY uploaded_at DESC LIMIT ?",
            (limit,),
        )
        return [_video_from_row(row) for row in rows]

    async def get_setting_updated_at(self, key: str) -> str | None:
        row = await self.db.fetch_one("SELECT updated_at FROM settings WHERE key = ?", (key,))
        return row["updated_at"] if row else None