"""
SQLite persistence layer.

All public methods are `async` and delegate the blocking sqlite3 work to a
worker thread (asyncio.to_thread) so the Telegram event loop never blocks.
Every SQL statement uses parameterized queries.
"""

from __future__ import annotations

import asyncio
import logging
import sqlite3
import threading
from pathlib import Path
from typing import Any, Optional

import utils

logger = logging.getLogger(__name__)

_SCHEMA = """
CREATE TABLE IF NOT EXISTS users (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    user_id INTEGER NOT NULL UNIQUE,
    username TEXT,
    first_name TEXT,
    joined_at TEXT,
    last_active TEXT,
    total_downloads INTEGER NOT NULL DEFAULT 0,
    is_banned INTEGER NOT NULL DEFAULT 0
);

CREATE TABLE IF NOT EXISTS downloads (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    user_id INTEGER NOT NULL,
    url TEXT NOT NULL,
    title TEXT,
    format TEXT,
    file_size INTEGER,
    status TEXT NOT NULL DEFAULT 'started',
    created_at TEXT
);

CREATE TABLE IF NOT EXISTS force_join (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    chat_id TEXT NOT NULL,
    username TEXT,
    title TEXT,
    invite_link TEXT,
    chat_type TEXT,
    is_active INTEGER NOT NULL DEFAULT 1
);

CREATE TABLE IF NOT EXISTS settings (
    key TEXT PRIMARY KEY,
    value TEXT
);

CREATE TABLE IF NOT EXISTS banned_users (
    user_id INTEGER PRIMARY KEY,
    reason TEXT,
    created_at TEXT
);

CREATE INDEX IF NOT EXISTS idx_downloads_user_id ON downloads (user_id);
CREATE INDEX IF NOT EXISTS idx_downloads_status ON downloads (status);
CREATE INDEX IF NOT EXISTS idx_users_last_active ON users (last_active);
"""


class Database:
    """Async facade over a single sqlite3 connection guarded by a lock."""

    def __init__(self, db_path: Path) -> None:
        self._path = Path(db_path)
        self._lock = threading.RLock()
        self._conn: Optional[sqlite3.Connection] = None

    # ------------------------------------------------------------------ #
    # Lifecycle
    # ------------------------------------------------------------------ #

    async def initialize(self) -> None:
        """Connect and create the schema if it does not exist yet."""
        await asyncio.to_thread(self._connect)

    def _connect(self) -> None:
        self._path.parent.mkdir(parents=True, exist_ok=True)
        conn = sqlite3.connect(str(self._path), check_same_thread=False)
        conn.row_factory = sqlite3.Row
        self._conn = conn
        with self._lock:
            self._conn.executescript(_SCHEMA)
            self._conn.commit()
        logger.info("Database ready at %s", self._path)

    async def close(self) -> None:
        await asyncio.to_thread(self._close)

    def _close(self) -> None:
        with self._lock:
            if self._conn is not None:
                self._conn.close()
                self._conn = None

    # ------------------------------------------------------------------ #
    # Low-level helpers
    # ------------------------------------------------------------------ #

    def _execute(self, sql: str, params: tuple = ()) -> sqlite3.Cursor:
        if self._conn is None:
            raise RuntimeError("Database is not initialized")
        with self._lock:
            try:
                cursor = self._conn.execute(sql, params)
                self._conn.commit()
                return cursor
            except sqlite3.Error:
                logger.exception("Database error while executing: %s", sql)
                raise

    def _fetchone(self, sql: str, params: tuple = ()) -> Optional[sqlite3.Row]:
        if self._conn is None:
            raise RuntimeError("Database is not initialized")
        with self._lock:
            return self._conn.execute(sql, params).fetchone()

    def _fetchall(self, sql: str, params: tuple = ()) -> list[sqlite3.Row]:
        if self._conn is None:
            raise RuntimeError("Database is not initialized")
        with self._lock:
            return self._conn.execute(sql, params).fetchall()

    @staticmethod
    def _d(row: Optional[sqlite3.Row]) -> Optional[dict[str, Any]]:
        return dict(row) if row is not None else None

    # ------------------------------------------------------------------ #
    # Users
    # ------------------------------------------------------------------ #

    async def upsert_user(self, user_id: int, username: Optional[str], first_name: Optional[str]) -> None:
        await asyncio.to_thread(self._upsert_user, user_id, username, first_name)

    def _upsert_user(self, user_id: int, username: Optional[str], first_name: Optional[str]) -> None:
        now = utils.now_str()
        self._execute(
            """
            INSERT INTO users (user_id, username, first_name, joined_at, last_active)
            VALUES (?, ?, ?, ?, ?)
            ON CONFLICT(user_id) DO UPDATE SET
                username = excluded.username,
                first_name = excluded.first_name,
                last_active = excluded.last_active
            """,
            (user_id, username, first_name, now, now),
        )

    async def touch_user(self, user_id: int) -> None:
        await asyncio.to_thread(self._touch_user, user_id)

    def _touch_user(self, user_id: int) -> None:
        self._execute("UPDATE users SET last_active = ? WHERE user_id = ?", (utils.now_str(), user_id))

    async def get_user(self, user_id: int) -> Optional[dict[str, Any]]:
        row = await asyncio.to_thread(
            self._fetchone, "SELECT * FROM users WHERE user_id = ?", (user_id,)
        )
        return self._d(row)

    async def recent_users(self, limit: int = 10) -> list[dict[str, Any]]:
        rows = await asyncio.to_thread(
            self._fetchall,
            "SELECT * FROM users ORDER BY last_active DESC LIMIT ?",
            (limit,),
        )
        return [dict(r) for r in rows]

    async def count_users(self) -> int:
        row = await asyncio.to_thread(self._fetchone, "SELECT COUNT(*) AS c FROM users")
        return int(row["c"]) if row else 0

    async def all_user_ids(self) -> list[int]:
        rows = await asyncio.to_thread(self._fetchall, "SELECT user_id FROM users")
        return [int(r["user_id"]) for r in rows]

    async def increment_user_downloads(self, user_id: int) -> None:
        await asyncio.to_thread(self._increment_user_downloads, user_id)

    def _increment_user_downloads(self, user_id: int) -> None:
        self._execute(
            "UPDATE users SET total_downloads = total_downloads + 1 WHERE user_id = ?",
            (user_id,),
        )

    async def user_stats(self, user_id: int) -> Optional[dict[str, Any]]:
        return await asyncio.to_thread(self._user_stats, user_id)

    def _user_stats(self, user_id: int) -> Optional[dict[str, Any]]:
        row = self._fetchone(
            "SELECT total_downloads, joined_at FROM users WHERE user_id = ?", (user_id,)
        )
        if row is None:
            return None
        count = self._fetchone(
            "SELECT COUNT(*) AS c FROM downloads WHERE user_id = ?", (user_id,)
        )
        return {
            "total_downloads": int(row["total_downloads"]),
            "joined_at": row["joined_at"],
            "downloads_count": int(count["c"]) if count else 0,
        }

    # ------------------------------------------------------------------ #
    # Downloads
    # ------------------------------------------------------------------ #

    async def add_download(self, user_id: int, url: str, title: Optional[str], fmt: Optional[str]) -> int:
        return await asyncio.to_thread(self._add_download, user_id, url, title, fmt)

    def _add_download(self, user_id: int, url: str, title: Optional[str], fmt: Optional[str]) -> int:
        cursor = self._execute(
            """
            INSERT INTO downloads (user_id, url, title, format, status, created_at)
            VALUES (?, ?, ?, ?, 'started', ?)
            """,
            (user_id, url, title, fmt, utils.now_str()),
        )
        return int(cursor.lastrowid or 0)

    async def finish_download(self, download_id: int, status: str, file_size: Optional[int]) -> None:
        await asyncio.to_thread(self._finish_download, download_id, status, file_size)

    def _finish_download(self, download_id: int, status: str, file_size: Optional[int]) -> None:
        self._execute(
            "UPDATE downloads SET status = ?, file_size = COALESCE(?, file_size) WHERE id = ?",
            (status, file_size, download_id),
        )

    async def stats_overview(self) -> dict[str, int]:
        return await asyncio.to_thread(self._stats_overview)

    def _stats_overview(self) -> dict[str, int]:
        def count(sql: str, params: tuple = ()) -> int:
            row = self._fetchone(sql, params)
            return int(row["c"]) if row else 0

        return {
            "total_users": count("SELECT COUNT(*) AS c FROM users"),
            "total_downloads": count("SELECT COUNT(*) AS c FROM downloads"),
            "completed": count("SELECT COUNT(*) AS c FROM downloads WHERE status = 'completed'"),
            "failed": count("SELECT COUNT(*) AS c FROM downloads WHERE status = 'failed'"),
            "cancelled": count("SELECT COUNT(*) AS c FROM downloads WHERE status = 'cancelled'"),
            "banned": count("SELECT COUNT(*) AS c FROM banned_users"),
            "active_week": count(
                "SELECT COUNT(*) AS c FROM users WHERE last_active >= ?", (utils.past_date_str(7),)
            ),
            "force_join": count("SELECT COUNT(*) AS c FROM force_join WHERE is_active = 1"),
        }

    # ------------------------------------------------------------------ #
    # Bans
    # ------------------------------------------------------------------ #

    async def ban_user(self, user_id: int, reason: str) -> None:
        await asyncio.to_thread(self._ban_user, user_id, reason)

    def _ban_user(self, user_id: int, reason: str) -> None:
        self._execute(
            "INSERT OR REPLACE INTO banned_users (user_id, reason, created_at) VALUES (?, ?, ?)",
            (user_id, reason, utils.now_str()),
        )
        self._execute("UPDATE users SET is_banned = 1 WHERE user_id = ?", (user_id,))

    async def unban_user(self, user_id: int) -> None:
        await asyncio.to_thread(self._unban_user, user_id)

    def _unban_user(self, user_id: int) -> None:
        self._execute("DELETE FROM banned_users WHERE user_id = ?", (user_id,))
        self._execute("UPDATE users SET is_banned = 0 WHERE user_id = ?", (user_id,))

    async def is_user_banned(self, user_id: int) -> bool:
        row = await asyncio.to_thread(
            self._fetchone, "SELECT is_banned FROM users WHERE user_id = ?", (user_id,)
        )
        return bool(row and int(row["is_banned"]) == 1)

    async def get_ban_reason(self, user_id: int) -> Optional[str]:
        row = await asyncio.to_thread(
            self._fetchone, "SELECT reason FROM banned_users WHERE user_id = ?", (user_id,)
        )
        return row["reason"] if row else None

    # ------------------------------------------------------------------ #
    # Settings (key/value)
    # ------------------------------------------------------------------ #

    async def get_setting(self, key: str, default: Optional[str] = None) -> Optional[str]:
        row = await asyncio.to_thread(self._fetchone, "SELECT value FROM settings WHERE key = ?", (key,))
        if row is None or row["value"] is None:
            return default
        return str(row["value"])

    async def set_setting(self, key: str, value: str) -> None:
        await asyncio.to_thread(self._set_setting, key, value)

    def _set_setting(self, key: str, value: str) -> None:
        self._execute(
            "INSERT INTO settings (key, value) VALUES (?, ?) "
            "ON CONFLICT(key) DO UPDATE SET value = excluded.value",
            (key, value),
        )

    async def get_setting_int(self, key: str, default: int) -> int:
        raw = await self.get_setting(key)
        if raw is None:
            return default
        try:
            return int(raw)
        except (TypeError, ValueError):
            return default

    # ------------------------------------------------------------------ #
    # Force join
    # ------------------------------------------------------------------ #

    async def add_force_join(
        self,
        chat_id: str,
        username: Optional[str],
        title: Optional[str],
        invite_link: Optional[str],
        chat_type: Optional[str],
    ) -> int:
        return await asyncio.to_thread(
            self._add_force_join, chat_id, username, title, invite_link, chat_type
        )

    def _add_force_join(
        self,
        chat_id: str,
        username: Optional[str],
        title: Optional[str],
        invite_link: Optional[str],
        chat_type: Optional[str],
    ) -> int:
        cursor = self._execute(
            """
            INSERT INTO force_join (chat_id, username, title, invite_link, chat_type, is_active)
            VALUES (?, ?, ?, ?, ?, 1)
            """,
            (chat_id, username, title, invite_link, chat_type),
        )
        return int(cursor.lastrowid or 0)

    async def list_force_join(self, active_only: bool = True) -> list[dict[str, Any]]:
        sql = "SELECT * FROM force_join"
        if active_only:
            sql += " WHERE is_active = 1"
        sql += " ORDER BY id"
        rows = await asyncio.to_thread(self._fetchall, sql)
        return [dict(r) for r in rows]

    async def get_force_join(self, entry_id: int) -> Optional[dict[str, Any]]:
        row = await asyncio.to_thread(self._fetchone, "SELECT * FROM force_join WHERE id = ?", (entry_id,))
        return self._d(row)

    async def remove_force_join(self, entry_id: int) -> None:
        await asyncio.to_thread(self._execute, "DELETE FROM force_join WHERE id = ?", (entry_id,))

    async def set_force_join_active(self, entry_id: int, active: bool) -> None:
        await asyncio.to_thread(
            self._execute, "UPDATE force_join SET is_active = ? WHERE id = ?", (1 if active else 0, entry_id)
        )

    async def count_force_join(self, active_only: bool = True) -> int:
        sql = "SELECT COUNT(*) AS c FROM force_join"
        if active_only:
            sql += " WHERE is_active = 1"
        row = await asyncio.to_thread(self._fetchone, sql)
        return int(row["c"]) if row else 0