# -*- coding: utf-8 -*-
"""路径常量、连接与建表（原生异步 SQLite 访问层：aiosqlite + WAL）。

（自 auth_system.py 拆出；本文件由同步 sqlite3 栈迁移为原生异步，设计照
`core/db.py`——同一套「读写同时进行」口径：）

- WAL 模式：读事务与写事务互不阻塞（读快照隔离）；
- 读路径（`_q`）：每次操作独立连接、无锁、可并发；
- 写路径（`_x` / `write_transaction`）：SQLite 全库同一时刻只允许一个写者，
  用 `asyncio.Lock` 串行化写事务，但写者不阻塞读者；
- `busy_timeout`：多进程/多连接下遇短暂写锁自旋等待而非立刻 SQLITE_BUSY。

路径常量仍是本模块的**唯一真源**：DB 访问函数在**调用时**读取本模块的
`DB_PATH` 等全局（测试 monkeypatch `auth_system.db.DB_PATH` 依旧生效）。
"""

import asyncio
import os
import secrets
import sqlite3
import sys
import time
from collections.abc import AsyncGenerator, Sequence
from contextlib import asynccontextmanager
from pathlib import Path

import aiosqlite

from .security import (
    _hash_new,
    _normalize_email,
)

# ============ 路径与常量 ============
# 必须与 core.db.DATA_DIR 用**同一条**表达式：两处指向同一个 data/site.db。
# 此前这里写死 Path('data') 而 core 侧认 SITE_DATA_DIR——一旦运维设置该变量，
# 站点库与设置/公告库会落到两个不同文件（脑裂），且不会报错。
DATA_DIR = Path(os.environ.get('SITE_DATA_DIR') or 'data')

DB_PATH = DATA_DIR / 'site.db'

BACKUP_DIR = DATA_DIR / 'backups'

SECRET_KEY_PATH = DATA_DIR / 'secret.key'


DEFAULT_ADMIN_USER = 'admin'

SESSION_KEY = 'site_uid'

DEFAULT_REGISTER_CREDITS = 3      # 注册直接赠送 3 次下载次数（无需领取；TRIAL_CREDITS 死常量已删）


# 锁按事件循环隔离：asyncio.Lock 绑定创建它的 loop，跨 loop 复用会抛
# "bound to a different event loop"（应用主 loop + 测试各自的 loop 并存，必须分开持有）。
_LOOP_LOCKS: dict[tuple[int, str], asyncio.Lock] = {}


def _loop_lock(kind: str) -> asyncio.Lock:
    key = (id(asyncio.get_running_loop()), kind)
    lock = _LOOP_LOCKS.get(key)
    if lock is None:
        lock = asyncio.Lock()
        _LOOP_LOCKS[key] = lock
    return lock


def _write_lock() -> asyncio.Lock:
    """写事务串行锁（每事件循环一个）。"""
    return _loop_lock('write')


def _wal_lock() -> asyncio.Lock:
    """WAL 切换专用锁（不能用 _write_lock：execute() 已持它再调
    connect→_configure 会自锁死）。"""
    return _loop_lock('wal')


# journal_mode=WAL 持久化在库文件头里，按路径设一次即可。每连接重复执行
# 实测多付 ~1ms（新连接首条语句要建 WAL 索引读锁），且并发首连会争排他锁
# 报 database is locked（core/db.py 同款教训）。专用锁 + 二次复查防护。
_WAL_READY_FOR: str | None = None


# ============ 数据库 ============


async def _configure(conn: aiosqlite.Connection, force_wal: bool = False) -> None:
    global _WAL_READY_FOR
    conn.row_factory = aiosqlite.Row
    # busy_timeout 先设：让后续语句遇锁自旋等待，而不是立刻报 database is locked
    await conn.execute('PRAGMA busy_timeout=15000')
    key = str(DB_PATH)   # 调用时读取模块全局——测试 monkeypatch 后仍生效
    if force_wal or _WAL_READY_FOR != key:
        # 切换日志模式需要排他锁，且 `_WAL_READY_FOR` 的检查不是原子的——
        # 并发连接须在专用锁内串行完成「切换 + 标记」，并二次复查，否则批量并发
        # 连接会同时执行 PRAGMA journal_mode=WAL 争锁（实测 database is locked）
        async with _wal_lock():
            if force_wal or _WAL_READY_FOR != key:
                await conn.execute('PRAGMA journal_mode=WAL')
                _WAL_READY_FOR = key
    await conn.execute('PRAGMA synchronous=NORMAL')  # WAL 下兼顾安全与吞吐
    await conn.execute('PRAGMA foreign_keys=ON')


async def connect() -> aiosqlite.Connection:
    """新建一个已配置好的连接（调用方负责 close）。"""
    # exists() 必须在 aiosqlite.connect 之前判断：connect 会创建空文件，之后
    # exists 恒为 True，库文件被删重建的场景就检测不到了（journal_mode 不持久
    # 于新文件，必须重设）。
    force_wal = not DB_PATH.exists()
    # mkdir 移出 happy path（实测 190us/次）：connect 直接失败（目录缺失）才补建
    # 重试。库文件被删的场景同样由此兜底重建。用 DB_PATH.parent 而非 DATA_DIR：
    # 生产两者相同，测试 monkeypatch 只换 DB_PATH 时兜底仍能命中正确目录
    # （core/db.py 用 DATA_DIR 的前提是两处都换）。
    try:
        conn = await aiosqlite.connect(str(DB_PATH), timeout=15)
    except sqlite3.OperationalError:
        DB_PATH.parent.mkdir(parents=True, exist_ok=True)
        conn = await aiosqlite.connect(str(DB_PATH), timeout=15)
    await _configure(conn, force_wal)
    return conn


# 注：同步时代的公开连接门面 get_conn() 已删除——它在迁移后只剩「转调 connect()」
# 一层空壳且全仓零消费者（测试直接开裸连接，备份用 db.connect()）。保留它等于留一条
# 「拿裸连接自己写」的口子，与本次「拆掉同步门面、写路径统一走 _x/write_transaction」
# 的目标相悖。


# ============ 读路径（并发，无锁） ============


async def _q(sql: str, params: Sequence = ()) -> list[aiosqlite.Row]:
    """单语句读：**不持锁**。

    WAL 模式下每个连接读到一致性快照，读与写互不阻塞；旧实现对读也加锁，
    等于把全部读串行化并被写挡住。此处仅放开单语句读——写事务（_x 与各
    `write_transaction` 块）仍全程串行，故不引入写-写竞争。
    """
    conn = await connect()
    try:
        cur = await conn.execute(sql, params)
        return list(await cur.fetchall())
    finally:
        await conn.close()


# ============ 写路径（串行，不阻塞读） ============


async def _x(sql: str, params: Sequence = ()) -> int:
    """执行单条写语句，返回 lastrowid（INSERT）或 rowcount（UPDATE/DELETE）"""
    async with _write_lock():
        conn = await connect()
        try:
            cur = await conn.execute(sql, params)
            await conn.commit()
            return cur.lastrowid if cur.lastrowid else cur.rowcount
        finally:
            await conn.close()


@asynccontextmanager
async def write_transaction() -> AsyncGenerator[aiosqlite.Connection, None]:
    """写事务上下文：多条写语句原子提交（串行锁内）。

    用法：
        async with write_transaction() as conn:
            await conn.execute(...)
            await conn.execute(...)
        # 退出即 commit；异常则 rollback
    """
    async with _write_lock():
        conn = await connect()
        try:
            yield conn
            await conn.commit()
        except Exception:
            await conn.rollback()
            raise
        finally:
            await conn.close()


_SCHEMA_SQL = '''
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        username TEXT UNIQUE NOT NULL,
        password_hash TEXT NOT NULL,
        salt TEXT NOT NULL,
        email TEXT DEFAULT '',
        role TEXT DEFAULT 'user',
        duration_seconds INTEGER DEFAULT 0,
        trial_device TEXT DEFAULT '',
        register_ip TEXT DEFAULT '',
        register_ua TEXT DEFAULT '',
        netease_cookie TEXT DEFAULT '',
        ann_seen TEXT DEFAULT '',
        last_active INTEGER DEFAULT 0,
        created_at INTEGER DEFAULT 0,
        last_login INTEGER DEFAULT 0
    );
    CREATE TABLE IF NOT EXISTS email_codes (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        email TEXT NOT NULL,
        code TEXT NOT NULL,
        expires_at INTEGER NOT NULL,
        created_at INTEGER DEFAULT 0,
        purpose TEXT DEFAULT 'register'
    );
    CREATE TABLE IF NOT EXISTS cards (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        code TEXT UNIQUE NOT NULL,
        minutes INTEGER NOT NULL,
        created_by TEXT DEFAULT '',
        created_at INTEGER DEFAULT 0,
        used_by TEXT DEFAULT '',
        used_at INTEGER DEFAULT 0
    );
    CREATE TABLE IF NOT EXISTS logs (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        type TEXT DEFAULT 'success',
        username TEXT DEFAULT '',
        ip TEXT DEFAULT '',
        action TEXT DEFAULT '',
        detail TEXT DEFAULT '',
        time TEXT DEFAULT ''
    );
'''


async def _create_schema(conn: aiosqlite.Connection) -> None:
    """建表 + 单行配置种子行（幂等：全部 IF NOT EXISTS / INSERT OR IGNORE）。"""
    await conn.executescript(_SCHEMA_SQL)
    # 单行配置表
    await conn.execute('''CREATE TABLE IF NOT EXISTS smtp_config (
        id INTEGER PRIMARY KEY CHECK (id = 1),
        host TEXT DEFAULT '', port INTEGER DEFAULT 587,
        sender TEXT DEFAULT '', password TEXT DEFAULT '',
        updated_at INTEGER DEFAULT 0
    )''')
    await conn.execute('INSERT OR IGNORE INTO smtp_config (id, port) VALUES (1, 587)')
    # 键值设置表
    await conn.execute('''CREATE TABLE IF NOT EXISTS settings (
        key TEXT PRIMARY KEY, value TEXT DEFAULT ''
    )''')
    await conn.execute('''CREATE TABLE IF NOT EXISTS announcements (
        id INTEGER PRIMARY KEY CHECK (id = 1),
        title TEXT DEFAULT '', content TEXT DEFAULT '',
        popup_mode TEXT DEFAULT 'off',
        updated_at INTEGER DEFAULT 0
    )''')
    await conn.execute('INSERT OR IGNORE INTO announcements (id) VALUES (1)')
    # 计费账本（任务实扣/交付/退款/终态的持久化唯一记录，结算幂等）
    await conn.execute('''CREATE TABLE IF NOT EXISTS billing_ledger (
        task_id TEXT PRIMARY KEY,
        user_id INTEGER NOT NULL,
        reserved INTEGER DEFAULT 0,
        delivered INTEGER DEFAULT 0,
        refunded INTEGER DEFAULT 0,
        status TEXT DEFAULT 'open',
        created_at INTEGER DEFAULT 0,
        settled_at INTEGER DEFAULT 0
    )''')


async def _apply_column_migrations(conn: aiosqlite.Connection) -> None:
    """轻量迁移：新增列。仅"列已存在"可忽略（其余迁移失败必须暴露，不能吞掉）"""
    _already: set[str] = set()
    trows = await (await conn.execute("SELECT name FROM sqlite_master WHERE type='table'")).fetchall()
    for trow in trows:
        crows = await (await conn.execute(f"PRAGMA table_info({trow['name']})")).fetchall()
        for crow in crows:
            _already.add(f"{trow['name']}.{crow['name']}")
    for alter_sql, tbl_col in (
        ('ALTER TABLE users ADD COLUMN status INTEGER DEFAULT 1', 'users.status'),
        ('ALTER TABLE users ADD COLUMN login_seq INTEGER DEFAULT 0', 'users.login_seq'),
        ('ALTER TABLE users ADD COLUMN recent_provinces TEXT DEFAULT \'[]\'', 'users.recent_provinces'),
        ('ALTER TABLE users ADD COLUMN ban_reason TEXT DEFAULT \'\'', 'users.ban_reason'),
        ('ALTER TABLE cards ADD COLUMN note TEXT DEFAULT \'\'', 'cards.note'),
        ('ALTER TABLE cards ADD COLUMN expire_at INTEGER DEFAULT 0', 'cards.expire_at'),
        ('ALTER TABLE cards ADD COLUMN disabled INTEGER DEFAULT 0', 'cards.disabled'),
        ('ALTER TABLE email_codes ADD COLUMN purpose TEXT DEFAULT \'register\'', 'email_codes.purpose'),
    ):
        if tbl_col in _already:
            continue  # 幂等：列已存在
        try:
            await conn.execute(alter_sql)
        except sqlite3.OperationalError as e:
            if 'duplicate column' not in str(e).lower():
                # 磁盘/锁/结构等真实故障：抛出，禁止静默带病初始化
                raise


async def _ensure_email_unique_index(conn: aiosqlite.Connection) -> None:
    """非空邮箱唯一（部分索引，不约束历史空串）。

    存量库若已有重复非空邮箱，索引创建失败仅告警不阻断启动
    （查重逻辑在注册/建号路由另有兜底）。
    """
    try:
        await conn.execute('CREATE UNIQUE INDEX IF NOT EXISTS idx_users_email_nonempty '
                           'ON users(email) WHERE email != \'\'')
    except sqlite3.IntegrityError:
        print('[警告] users.email 存在重复的非空邮箱，唯一索引未创建（请人工清理后重启）')


async def _migrate_user_emails(conn: aiosqlite.Connection) -> None:
    """users.email 口径统一为归一化（lower + 剥 Gmail +tag）——与
    get_user_by_email/找回/邮箱登录的查询口径对齐。逐行迁移，归一化后与他
    人撞唯一索引的行保留原值跳过（与"存量重复仅告警"同口径）。幂等：已
    归一化的行零写入。
    """
    urows = await (await conn.execute("SELECT id, email FROM users WHERE email != ''")).fetchall()
    for urow in urows:
        fixed = _normalize_email(urow['email'])
        if not fixed or fixed == urow['email']:
            continue
        try:
            await conn.execute('UPDATE users SET email = ? WHERE id = ?', (fixed, urow['id']))
        except sqlite3.IntegrityError:
            print(f"[警告] 用户 {urow['id']} 的邮箱归一化后与他人重复，保留原值")


async def _ensure_initial_admin(conn: aiosqlite.Connection) -> None:
    """初始管理员（新装不再使用公开默认口令，随机生成并写入日志一次）。

    引导条件改为"不存在任何管理员角色"——按用户名查会在
    管理员改名（如 admin→qfbsh）后误判缺失，重启即重建出一个新的 admin 账号。
    """
    row = await (await conn.execute("SELECT id FROM users WHERE role = 'admin' LIMIT 1")).fetchone()
    if row:
        return
    initial_pass = secrets.token_urlsafe(12)
    await conn.execute(
        'INSERT INTO users (username, password_hash, salt, role, duration_seconds, created_at) '
        'VALUES (?, ?, ?, ?, -1, ?)',
        (DEFAULT_ADMIN_USER, _hash_new(initial_pass), '',
         'admin', int(time.time())))
    # ——非 ASCII 输出须防 GBK 重定向下的 UnicodeEncodeError
    try:
        print(f'[初始化] 已创建管理员账号 admin，初始密码：{initial_pass}（请立即登录并修改，本提示不再显示）')
    except UnicodeEncodeError:
        print('[init] admin account created, initial password printed above failed to encode')


async def _init_db_locked(conn: aiosqlite.Connection) -> None:
    """在**已持有写锁**的连接上执行全部结构演进（不自行取锁）。

    restore_backup 需要在自己的写锁临界区里对新库跑一遍初始化，不能调用会
    再次取写锁的 init_db（asyncio.Lock 非重入，会自锁死），故拆出本函数。
    """
    await _create_schema(conn)
    await _apply_column_migrations(conn)
    await _ensure_email_unique_index(conn)
    await _migrate_user_emails(conn)
    await _ensure_initial_admin(conn)


async def init_db() -> None:
    """初始化表结构与默认数据"""
    DATA_DIR.mkdir(parents=True, exist_ok=True)  # noqa: ASYNC240 - 启动期一次的目录创建，无需线程池
    async with write_transaction() as conn:
        await _init_db_locked(conn)


def _get_secret_key_sync() -> str:
    """同步实现（见 get_secret_key 的异步包装）：原子写入 + 失败显式告警"""
    key = ''
    try:
        if SECRET_KEY_PATH.exists():
            key = SECRET_KEY_PATH.read_text(encoding='utf-8').strip()
            if key:
                return key
    except OSError:
        pass
    new_key = secrets.token_hex(32)
    try:
        SECRET_KEY_PATH.parent.mkdir(parents=True, exist_ok=True)
        # 先写临时文件再原子替换，避免部分写入留下损坏密钥
        tmp = SECRET_KEY_PATH.with_suffix('.key.tmp')
        tmp.write_text(new_key, encoding='utf-8')
        os.replace(str(tmp), str(SECRET_KEY_PATH))
        # 并发初始化者以"替换后文件里的值"为准（最后写入者胜出但全员一致）
        final = SECRET_KEY_PATH.read_text(encoding='utf-8').strip()
        return final or new_key
    except OSError as e:
        # 写入失败不能装作成功：显著告警，返回临时值（本进程会话可用，重启后失效）
        try:
            print(f'[警告] 会话密钥持久化失败（{e}），本进程将使用临时密钥，重启后所有会话失效！', file=sys.stderr)
        except UnicodeEncodeError:
            print('[warn] secret key persistence failed; using temporary key for this process', file=sys.stderr)
        return new_key


async def get_secret_key() -> str:
    """获取（或生成）持久化的会话签名密钥（原子写入 + 失败显式告警）。

    内部是**同步阻塞文件 IO**（exists/read/write/os.replace），整段经
    `asyncio.to_thread` 卸载到线程——这是正当的阻塞 IO 卸载（登录链路每次
    签会话都要读一次密钥），不是把异步包装成同步的门面。
    """
    return await asyncio.to_thread(_get_secret_key_sync)