"""DB adapter for SQLite."""

import sqlite3
from pathlib import Path
from typing import Any

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import (
    create_engine as sa_create_engine,
    event,
    exists,
    func,
    literal,
    select,
)
from sqlalchemy.engine import URL, Connection, Engine
from sqlalchemy.pool import StaticPool
from sqlalchemy.sql.elements import ClauseElement

from .adapter import Adapter
from .utils import DialectName


class SQLiteConfig:
    """SQLite Config."""

    SQLITE_DB_NAME: str
    SQLITE_DIR: str
    SQLITE_EXT: str


class SQLiteAdapter(Adapter):
    """DB adapter for SQLite."""

    dialect = DialectName('sqlite')

    def clone_db(self, src_engine: Engine, dst_engine: Engine) -> None:
        """Clone the entire SQLite database from src_engine into dst_engine."""
        # Sanity checks
        if src_engine.dialect.name != 'sqlite' or dst_engine.dialect.name != 'sqlite':
            raise ValueError('Only SQLite engines supported.')

        # Avoid cloning onto the same underlying database
        src_db = src_engine.url.database
        dst_db = dst_engine.url.database
        if src_db and dst_db and src_db == dst_db:
            raise ValueError('Source and destination are the same database.')

        # Get raw sqlite3 connections from the SQLAlchemy engines
        src_raw, src_dbapi = _get_dbapi(src_engine)
        dst_raw, dst_dbapi = _get_dbapi(dst_engine)
        try:
            # Put in autocommit in case destination is mid-transaction
            original_iso = dst_dbapi.isolation_level
            dst_dbapi.isolation_level = None  # autocommit mode
            try:
                # Copy everything at once
                src_dbapi.backup(dst_dbapi, pages=0)
            finally:
                # Restore isolation level
                dst_dbapi.isolation_level = original_iso
        finally:
            # Return connections to SA pool
            try:
                dst_raw.close()  # type: ignore
            finally:
                src_raw.close()  # type: ignore

    def create_db(self, config: SQLiteConfig) -> None:
        """Ensure the SQLite DB file exists for the given config."""
        url, is_memory = self._build_url(config)

        if is_memory:
            return

        # Make sure the file exists
        path = Path(url.database)
        path.parent.mkdir(parents=True, exist_ok=True)

        # Make sure the file is initialized
        with sqlite3.connect(str(path)) as conn:
            conn.execute('PRAGMA user_version = 0;')
            conn.commit()

    def create_engine(self, config: SQLiteConfig) -> Engine:
        """Create SQLite DB engine."""
        url, is_memory = self._build_url(config)
        is_test = _get_is_test(config)

        # Check if DB exists
        if not is_memory and not self.db_exists(config):
            raise ValueError('SQLite DB does not exist')

        # Use StaticPool to avoid WAL race conditions.
        engine_opts: dict[str, Any] = {
            'connect_args': {'check_same_thread': False},
            'pool_pre_ping': True,
            'poolclass': StaticPool,
        }

        engine = sa_create_engine(url, **engine_opts)
        self._attach_sqlite_pragmas(engine, is_memory=is_memory, is_test=is_test)

        return engine  # type: ignore

    def db_exists(self, config: SQLiteConfig, *, validate_header: bool = False) -> bool:
        """Check if the SQLite DB file exists."""
        url, is_memory = self._build_url(config)

        if is_memory:
            return True

        path_str = str(url.database)
        return _sqlite_path_exists(path_str, validate_header=validate_header)

    def json_contains(self, col, value, *, path: str = '$'):
        """Get the JSON_CONTAINS equivalent function."""
        je = func.json_each(col, literal(path)).table_valued('value')
        je_alias = je.alias('je')
        candidate = value if isinstance(value, ClauseElement) else literal(value)
        return exists(
            select(1).select_from(je_alias).where(je_alias.c.value == candidate)
        )

    def set_fk(self, conn: Connection, enable: bool = True) -> None:
        """Control foreign key constraints."""
        conn.exec_driver_sql('PRAGMA foreign_keys = %s' % ('ON' if enable else 'OFF'))

    def setup_db(self, db: SQLAlchemy, app: Flask, config: SQLiteConfig):
        """Set up the SQLite database."""
        url, is_memory = self._build_url(config)
        is_test = _get_is_test(config)

        app.config.setdefault('SQLALCHEMY_TRACK_MODIFICATIONS', False)
        app.config['SQLALCHEMY_DATABASE_URI'] = str(url)

        # Use StaticPool to avoid WAL race conditions.
        engine_opts: dict[str, Any] = {
            'connect_args': {'check_same_thread': False},
            'pool_pre_ping': True,
            'poolclass': StaticPool,
        }

        app.config['SQLALCHEMY_ENGINE_OPTIONS'] = engine_opts
        db.init_app(app)

        # Enforce foreign keys like MySQL
        with app.app_context():
            engine = db.engine

            # Avoid double-registration
            if not getattr(app, '_sqlite_pragmas_attached', False):
                self._attach_sqlite_pragmas(
                    engine, is_memory=is_memory, is_test=is_test
                )
                app._sqlite_pragmas_attached = True  # type: ignore

    def truncate_tables(self, conn: Connection, tables: list[str]) -> None:
        """Hard reset tables (and sequences) before a test transaction starts."""
        if not tables:
            return

        preparer = conn.dialect.identifier_preparer
        self.set_fk(conn, False)
        try:
            for t in tables:
                name = preparer.quote(t)
                conn.exec_driver_sql(f'DELETE FROM {name}')
            self._reset_sequences(conn, tables)
        finally:
            self.set_fk(conn, True)

        # Make sure changes are visible to all connections
        self.checkpoint_wal(conn)

    def checkpoint_wal(self, conn: Connection) -> None:
        """Force a WAL checkpoint."""
        try:
            # Ensures all connections see db changes
            conn.exec_driver_sql('PRAGMA wal_checkpoint(RESTART)')
        except Exception:
            # Ignore errors if not in WAL mode or other issues
            pass

    def _attach_sqlite_pragmas(
        self, engine: Engine, *, is_memory: bool, is_test: bool = False
    ) -> None:
        @event.listens_for(engine, 'connect')
        def _on_connect(dbapi_conn, _):
            cur = dbapi_conn.cursor()
            try:
                # Always applied
                cur.execute('PRAGMA busy_timeout=5000;')
                cur.execute('PRAGMA foreign_keys=ON;')
                cur.execute('PRAGMA temp_store=MEMORY;')

                if is_memory:
                    # In-memory settings
                    cur.execute('PRAGMA auto_vacuum=NONE;')
                else:
                    # File-based settings
                    cur.execute('PRAGMA cache_size=-200000;')  # ~200MB
                    cur.execute('PRAGMA journal_mode=WAL;')
                    cur.execute('PRAGMA mmap_size=268435456;')  # 256MB
                    cur.execute('PRAGMA synchronous=NORMAL;')
                    cur.execute('PRAGMA wal_autocheckpoint=100;')  # Pages

                if is_test:
                    # Test settings (faster, safe for temporary test data)
                    cur.execute('PRAGMA auto_vacuum=NONE;')
                    cur.execute('PRAGMA synchronous=OFF;')
            finally:
                cur.close()

    def _build_url(self, config: SQLiteConfig) -> tuple[URL, bool]:
        """Return (URL, is_memory). If name is ':memory:', use in-memory DB."""
        db_name = _normalize_filename(config.SQLITE_DB_NAME, config.SQLITE_EXT)

        if db_name == ':memory:':
            return URL.create('sqlite+pysqlite', database=':memory:'), True

        # Resolve relative path
        dir_path = Path(config.SQLITE_DIR)
        if not dir_path.is_absolute():
            dir_path = Path.cwd() / dir_path

        db_path = (dir_path / db_name).resolve()
        return URL.create('sqlite+pysqlite', database=str(db_path)), False

    def _reset_sequences(self, conn: Connection, seqs: list[str]) -> None:
        if not seqs:
            return

        has_seqs = conn.exec_driver_sql("""
            SELECT 1
            FROM sqlite_schema
            WHERE type='table' AND name='sqlite_sequence'
            LIMIT 1
        """).scalar()

        if not has_seqs:
            return

        for name in seqs:
            conn.exec_driver_sql(
                'DELETE FROM sqlite_sequence WHERE name = :name', {'name': name}
            )


def _get_dbapi(engine: Engine):
    raw = engine.raw_connection()
    try:
        dbapi = (
            getattr(raw, 'driver_connection', None)
            or getattr(raw, 'connection', None)
            or raw
        )
        if not isinstance(dbapi, sqlite3.Connection):
            raise TypeError(f'Expected sqlite3.Connection, got {type(dbapi)}')
        return raw, dbapi
    except Exception:
        raw.close()  # type: ignore
        raise


def _get_is_test(config: SQLiteConfig) -> bool:
    is_test = getattr(config, 'Testing', False)
    is_test = getattr(config, 'TESTING', is_test)
    return is_test


def _normalize_filename(name: str, ext: str) -> str:
    if name == ':memory:':
        return name
    ext = ext if ext.startswith('.') else f'.{ext}'
    return name if name.lower().endswith(ext.lower()) else f'{name}{ext}'


def _sqlite_path_exists(path_str: str, *, validate_header: bool = False) -> bool:
    p = Path(path_str)
    if not (p.exists() and p.is_file()):
        return False
    if not validate_header:
        return True
    try:
        with p.open('rb') as f:
            return f.read(16) == b'SQLite format 3\x00'
    except OSError:
        return False