SourceShareOSV1/server/db.py

import os
import sqlite3
import threading
from queue import LifoQueue
from typing import Any, Iterable, Optional

from flask import g
from werkzeug.security import generate_password_hash

from .context import get_config
from .core import isoformat, utcnow

try:
    import pymysql
    import pymysql.cursors
except Exception:
    pymysql = None

IntegrityError = sqlite3.IntegrityError

_mysql_pool_lock = threading.Lock()
_mysql_pool: LifoQueue | None = None
_active_backend_lock = threading.Lock()
_active_backend: str | None = None
_migrating_lock = threading.Lock()
_migrating = False
_db_switch_lock = threading.Lock()


def is_migrating() -> bool:
    with _migrating_lock:
        return bool(_migrating)


def _set_migrating(v: bool) -> None:
    global _migrating
    with _migrating_lock:
        _migrating = bool(v)


def reset_mysql_pool() -> None:
    global _mysql_pool
    with _mysql_pool_lock:
        pool = _mysql_pool
        _mysql_pool = None
    if pool is None:
        return
    while True:
        try:
            conn = pool.get_nowait()
        except Exception:
            break
        try:
            conn.close()
        except Exception:
            pass


def _sqlite_connect(path: str) -> sqlite3.Connection:
    conn = sqlite3.connect(path)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA foreign_keys=ON;")
    return conn


def get_ctl_db() -> sqlite3.Connection:
    db: Optional[sqlite3.Connection] = g.get("ctl_db")
    if db is not None:
        return db
    config = get_config()
    db = _sqlite_connect(str(config.database_path))
    g.ctl_db = db
    return db


def ctl_execute(sql: str, params: Iterable[Any] = ()) -> sqlite3.Cursor:
    db = get_ctl_db()
    cur = db.execute(sql, list(params))
    db.commit()
    return cur


def ctl_fetch_one(sql: str, params: Iterable[Any] = ()) -> Optional[sqlite3.Row]:
    db = get_ctl_db()
    cur = db.execute(sql, list(params))
    return cur.fetchone()


def ctl_fetch_all(sql: str, params: Iterable[Any] = ()) -> list[sqlite3.Row]:
    db = get_ctl_db()
    cur = db.execute(sql, list(params))
    return cur.fetchall()


def _ctl_ensure_schema() -> None:
    config = get_config()
    conn = _sqlite_connect(str(config.database_path))
    try:
        conn.executescript(
            """
            CREATE TABLE IF NOT EXISTS app_settings (
                key TEXT PRIMARY KEY,
                category TEXT NOT NULL DEFAULT 'GENERAL',
                value TEXT NOT NULL,
                updated_at TEXT NOT NULL
            );
            """
        )
        cols = []
        try:
            cols = [r["name"] for r in conn.execute("PRAGMA table_info(app_settings)").fetchall()]
        except Exception:
            cols = []
        if "category" not in cols:
            conn.execute("ALTER TABLE app_settings ADD COLUMN category TEXT NOT NULL DEFAULT 'GENERAL'")
        conn.commit()
    finally:
        conn.close()


def _ctl_get_value(key: str) -> str | None:
    config = get_config()
    conn = _sqlite_connect(str(config.database_path))
    try:
        try:
            row = conn.execute("SELECT value FROM app_settings WHERE key = ?", (key,)).fetchone()
        except Exception:
            return None
        if row is None:
            return None
        return (row["value"] or "").strip() or None
    finally:
        conn.close()


def _ctl_set_value(key: str, value: str, category: str = "GENERAL") -> None:
    now = isoformat(utcnow())
    config = get_config()
    conn = _sqlite_connect(str(config.database_path))
    try:
        conn.execute(
            """
            INSERT INTO app_settings (key, category, value, updated_at)
            VALUES (?, ?, ?, ?)
            ON CONFLICT(key) DO UPDATE SET
                category = excluded.category,
                value = excluded.value,
                updated_at = excluded.updated_at
            """,
            (key, category, value, now),
        )
        conn.commit()
    finally:
        conn.close()


def _ctl_delete_value(key: str) -> None:
    config = get_config()
    conn = _sqlite_connect(str(config.database_path))
    try:
        conn.execute("DELETE FROM app_settings WHERE key = ?", (key,))
        conn.commit()
    finally:
        conn.close()


def _resolve_backend_from_control() -> str | None:
    v = _ctl_get_value("DB_ACTIVE")
    if not v:
        return None
    v2 = v.strip().lower()
    if v2 in {"sqlite", "mysql"}:
        return v2
    return None


def get_active_backend() -> str:
    global _active_backend
    with _active_backend_lock:
        cached = _active_backend
    if cached in {"sqlite", "mysql"}:
        return cached
    v = _resolve_backend_from_control()
    with _active_backend_lock:
        _active_backend = v
        return v or "sqlite"


def set_active_backend(backend: str) -> None:
    b = (backend or "").strip().lower()
    if b not in {"sqlite", "mysql"}:
        b = "sqlite"
    with _active_backend_lock:
        global _active_backend
        prev = _active_backend
        _active_backend = b
    if prev != b and b == "mysql":
        reset_mysql_pool()


def get_db() -> sqlite3.Connection:
    db = g.get("db")
    if db is not None:
        return db

    config = get_config()
    backend = _resolve_backend_from_control()
    if backend is None:
        backend = "mysql" if (config.mysql_host or "").strip() else "sqlite"
    set_active_backend(backend)

    if backend == "mysql":
        host = (_ctl_get_value("MYSQL_HOST") or config.mysql_host or "").strip()
        user = (_ctl_get_value("MYSQL_USER") or config.mysql_user or "").strip()
        database = (_ctl_get_value("MYSQL_DATABASE") or config.mysql_database or "").strip()
        port = int((_ctl_get_value("MYSQL_PORT") or str(config.mysql_port or 3306)).strip() or 3306)
        password = (_ctl_get_value("MYSQL_PASSWORD") or config.mysql_password or "").strip()
        if not host or not user or not database:
            backend = "sqlite"
            set_active_backend("sqlite")
        else:
            if pymysql is None:
                raise RuntimeError("pymysql_required")
            global IntegrityError
            IntegrityError = getattr(pymysql.err, "IntegrityError", Exception)

            global _mysql_pool
            with _mysql_pool_lock:
                if _mysql_pool is None:
                    size = int(os.environ.get("MYSQL_POOL_SIZE", "10"))
                    _mysql_pool = LifoQueue(maxsize=max(1, min(size, 50)))
                pool = _mysql_pool

            conn = None
            try:
                conn = pool.get_nowait()
            except Exception:
                conn = None
            if conn is not None:
                try:
                    if not getattr(conn, "open", True):
                        raise RuntimeError("conn_closed")
                    conn.ping(reconnect=True)
                except Exception:
                    try:
                        conn.close()
                    except Exception:
                        pass
                    conn = None
            if conn is None:
                conn = pymysql.connect(
                    host=host,
                    port=port,
                    user=user,
                    password=password,
                    database=database,
                    charset="utf8mb4",
                    cursorclass=pymysql.cursors.DictCursor,
                    autocommit=False,
                )
            g.db = conn
            return conn

    db = sqlite3.connect(str(config.database_path))
    db.row_factory = sqlite3.Row
    db.execute("PRAGMA foreign_keys=ON;")
    g.db = db
    return db


def close_db(conn) -> None:
    if conn is None:
        return
    if _is_mysql(conn):
        global _mysql_pool
        pool = _mysql_pool
        if pool is None:
            try:
                conn.close()
            except Exception:
                pass
            return
        try:
            if getattr(conn, "open", True):
                conn.rollback()
        except Exception:
            try:
                conn.close()
            except Exception:
                pass
            return
        try:
            pool.put_nowait(conn)
        except Exception:
            try:
                conn.close()
            except Exception:
                pass
        return
    try:
        conn.close()
    except Exception:
        pass


def _is_mysql(conn) -> bool:
    if pymysql is None:
        return False
    return isinstance(conn, pymysql.connections.Connection)


def _translate_sql_for_mysql(sql: str) -> str:
    out: list[str] = []
    in_single = False
    in_double = False
    i = 0
    while i < len(sql):
        ch = sql[i]
        if ch == "'" and not in_double:
            out.append(ch)
            in_single = not in_single
            i += 1
            continue
        if ch == '"' and not in_single:
            out.append(ch)
            in_double = not in_double
            i += 1
            continue
        if ch == "?" and not in_single and not in_double:
            out.append("%s")
            i += 1
            continue
        out.append(ch)
        i += 1
    return "".join(out)


def execute(sql: str, params: Iterable[Any] = ()) -> sqlite3.Cursor:
    db = get_db()
    p = list(params)
    if _is_mysql(db):
        cur = db.cursor()
        cur.execute(_translate_sql_for_mysql(sql), p)
        db.commit()
        return cur
    cur = db.execute(sql, p)
    db.commit()
    return cur


def fetch_one(sql: str, params: Iterable[Any] = ()) -> Optional[sqlite3.Row]:
    db = get_db()
    p = list(params)
    if _is_mysql(db):
        cur = db.cursor()
        cur.execute(_translate_sql_for_mysql(sql), p)
        return cur.fetchone()
    cur = db.execute(sql, p)
    return cur.fetchone()


def fetch_all(sql: str, params: Iterable[Any] = ()) -> list[sqlite3.Row]:
    db = get_db()
    p = list(params)
    if _is_mysql(db):
        cur = db.cursor()
        cur.execute(_translate_sql_for_mysql(sql), p)
        return list(cur.fetchall() or [])
    cur = db.execute(sql, p)
    return cur.fetchall()


def init_db() -> None:
    _ctl_ensure_schema()
    db = get_db()
    if _is_mysql(db):
        stmts = [
            """
            CREATE TABLE IF NOT EXISTS users (
              id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
              phone VARCHAR(32) NOT NULL,
              password_hash VARCHAR(255) NOT NULL,
              status VARCHAR(16) NOT NULL DEFAULT 'ACTIVE',
              vip_expire_at VARCHAR(64) NULL,
              created_at VARCHAR(64) NOT NULL,
              UNIQUE KEY uk_users_phone (phone),
              KEY idx_users_status (status),
              KEY idx_users_vip_expire_at (vip_expire_at)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
            """,
            """
            CREATE TABLE IF NOT EXISTS admin_users (
              id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
              username VARCHAR(64) NOT NULL,
              password_hash VARCHAR(255) NOT NULL,
              role VARCHAR(16) NOT NULL DEFAULT 'ADMIN',
              status VARCHAR(16) NOT NULL DEFAULT 'ACTIVE',
              last_login_at VARCHAR(64) NULL,
              UNIQUE KEY uk_admin_users_username (username),
              KEY idx_admin_users_status (status)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
            """,
            """
            CREATE TABLE IF NOT EXISTS plans (
              id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
              name VARCHAR(120) NOT NULL,
              duration_days INT NOT NULL,
              price_cents INT NOT NULL,
              enabled TINYINT(1) NOT NULL DEFAULT 1,
              sort INT NOT NULL DEFAULT 0,
              KEY idx_plans_enabled_sort (enabled, sort)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
            """,
            """
            CREATE TABLE IF NOT EXISTS resources (
              id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
              title VARCHAR(200) NOT NULL,
              summary MEDIUMTEXT NOT NULL,
              type VARCHAR(8) NOT NULL,
              status VARCHAR(16) NOT NULL DEFAULT 'DRAFT',
              cover_url VARCHAR(500) NULL,
              category VARCHAR(120) NULL,
              tags_json TEXT NOT NULL,
              repo_owner VARCHAR(120) NOT NULL,
              repo_name VARCHAR(120) NOT NULL,
              repo_private TINYINT(1) NOT NULL DEFAULT 0,
              repo_html_url VARCHAR(500) NULL,
              default_ref VARCHAR(120) NOT NULL DEFAULT 'master',
              view_count INT NOT NULL DEFAULT 0,
              download_count INT NOT NULL DEFAULT 0,
              created_at VARCHAR(64) NOT NULL,
              updated_at VARCHAR(64) NOT NULL,
              KEY idx_resources_status_updated (status, updated_at),
              KEY idx_resources_status_views (status, view_count),
              KEY idx_resources_type_status (type, status)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
            """,
            """
            CREATE TABLE IF NOT EXISTS orders (
              id VARCHAR(64) NOT NULL PRIMARY KEY,
              user_id BIGINT UNSIGNED NOT NULL,
              plan_id BIGINT UNSIGNED NOT NULL,
              amount_cents INT NOT NULL,
              status VARCHAR(16) NOT NULL DEFAULT 'PENDING',
              pay_channel VARCHAR(32) NULL,
              pay_trade_no VARCHAR(128) NULL,
              created_at VARCHAR(64) NOT NULL,
              paid_at VARCHAR(64) NULL,
              plan_snapshot_json TEXT NOT NULL,
              KEY idx_orders_user_time (user_id, created_at),
              KEY idx_orders_status_time (status, created_at),
              KEY idx_orders_pay_trade_no (pay_trade_no)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
            """,
            """
            CREATE TABLE IF NOT EXISTS download_logs (
              id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
              user_id BIGINT UNSIGNED NOT NULL,
              resource_id BIGINT UNSIGNED NULL,
              resource_title_snapshot VARCHAR(200) NOT NULL,
              resource_type_snapshot VARCHAR(16) NOT NULL,
              ref_snapshot VARCHAR(200) NOT NULL,
              downloaded_at VARCHAR(64) NOT NULL,
              ip VARCHAR(64) NULL,
              user_agent VARCHAR(256) NULL,
              KEY idx_download_logs_user_time (user_id, downloaded_at),
              KEY idx_download_logs_resource (resource_id)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
            """,
            """
            CREATE TABLE IF NOT EXISTS audit_logs (
              id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
              actor_type VARCHAR(16) NOT NULL,
              actor_id BIGINT UNSIGNED NOT NULL,
              action VARCHAR(64) NOT NULL,
              resource_type VARCHAR(64) NOT NULL,
              resource_id VARCHAR(128) NOT NULL,
              before_json MEDIUMTEXT NULL,
              after_json MEDIUMTEXT NULL,
              ip VARCHAR(64) NULL,
              created_at VARCHAR(64) NOT NULL,
              KEY idx_audit_logs_time (created_at),
              KEY idx_audit_logs_actor (actor_type, actor_id)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
            """,
            """
            CREATE TABLE IF NOT EXISTS user_messages (
              id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
              user_id BIGINT UNSIGNED NOT NULL,
              title VARCHAR(120) NOT NULL,
              content MEDIUMTEXT NOT NULL,
              created_at VARCHAR(64) NOT NULL,
              read_at VARCHAR(64) NULL,
              sender_type VARCHAR(16) NOT NULL DEFAULT 'SYSTEM',
              sender_id BIGINT UNSIGNED NULL,
              KEY idx_user_messages_user_time (user_id, created_at),
              KEY idx_user_messages_user_read (user_id, read_at),
              KEY idx_user_messages_sender (sender_type, created_at)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
            """,
            """
            CREATE TABLE IF NOT EXISTS app_settings (
              `key` VARCHAR(200) NOT NULL PRIMARY KEY,
              category VARCHAR(32) NOT NULL DEFAULT 'GENERAL',
              value MEDIUMTEXT NOT NULL,
              updated_at VARCHAR(64) NOT NULL,
              KEY idx_app_settings_category (category)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
            """,
        ]
        for s in stmts:
            cur = db.cursor()
            cur.execute(s)
        db.commit()
        return

    db.executescript(
        """
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            phone TEXT NOT NULL UNIQUE,
            password_hash TEXT NOT NULL,
            status TEXT NOT NULL DEFAULT 'ACTIVE',
            vip_expire_at TEXT,
            created_at TEXT NOT NULL
        );
        CREATE INDEX IF NOT EXISTS idx_users_status ON users(status);
        CREATE INDEX IF NOT EXISTS idx_users_vip_expire_at ON users(vip_expire_at);

        CREATE TABLE IF NOT EXISTS admin_users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            username TEXT NOT NULL UNIQUE,
            password_hash TEXT NOT NULL,
            role TEXT NOT NULL DEFAULT 'ADMIN',
            status TEXT NOT NULL DEFAULT 'ACTIVE',
            last_login_at TEXT
        );
        CREATE INDEX IF NOT EXISTS idx_admin_users_status ON admin_users(status);

        CREATE TABLE IF NOT EXISTS plans (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL,
            duration_days INTEGER NOT NULL,
            price_cents INTEGER NOT NULL,
            enabled INTEGER NOT NULL DEFAULT 1,
            sort INTEGER NOT NULL DEFAULT 0
        );

        CREATE TABLE IF NOT EXISTS resources (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            title TEXT NOT NULL,
            summary TEXT NOT NULL DEFAULT '',
            type TEXT NOT NULL CHECK (type IN ('FREE', 'VIP')),
            status TEXT NOT NULL CHECK (status IN ('DRAFT', 'ONLINE', 'OFFLINE')) DEFAULT 'DRAFT',
            cover_url TEXT,
            category TEXT,
            tags_json TEXT NOT NULL DEFAULT '[]',
            repo_owner TEXT NOT NULL,
            repo_name TEXT NOT NULL,
            repo_private INTEGER NOT NULL DEFAULT 0,
            repo_html_url TEXT,
            default_ref TEXT NOT NULL DEFAULT 'master',
            view_count INTEGER NOT NULL DEFAULT 0,
            download_count INTEGER NOT NULL DEFAULT 0,
            created_at TEXT NOT NULL,
            updated_at TEXT NOT NULL
        );
        CREATE INDEX IF NOT EXISTS idx_resources_status_updated ON resources(status, updated_at);
        CREATE INDEX IF NOT EXISTS idx_resources_status_views ON resources(status, view_count);
        CREATE INDEX IF NOT EXISTS idx_resources_type_status ON resources(type, status);

        CREATE TABLE IF NOT EXISTS orders (
            id TEXT PRIMARY KEY,
            user_id INTEGER NOT NULL,
            plan_id INTEGER NOT NULL,
            amount_cents INTEGER NOT NULL,
            status TEXT NOT NULL CHECK (status IN ('PENDING', 'PAID', 'CLOSED', 'FAILED')) DEFAULT 'PENDING',
            pay_channel TEXT,
            pay_trade_no TEXT,
            created_at TEXT NOT NULL,
            paid_at TEXT,
            plan_snapshot_json TEXT NOT NULL,
            FOREIGN KEY(user_id) REFERENCES users(id),
            FOREIGN KEY(plan_id) REFERENCES plans(id)
        );
        CREATE INDEX IF NOT EXISTS idx_orders_user_time ON orders(user_id, created_at);
        CREATE INDEX IF NOT EXISTS idx_orders_status_time ON orders(status, created_at);
        CREATE INDEX IF NOT EXISTS idx_orders_pay_trade_no ON orders(pay_trade_no);

        CREATE TABLE IF NOT EXISTS download_logs (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            user_id INTEGER NOT NULL,
            resource_id INTEGER,
            resource_title_snapshot TEXT NOT NULL,
            resource_type_snapshot TEXT NOT NULL,
            ref_snapshot TEXT NOT NULL,
            downloaded_at TEXT NOT NULL,
            ip TEXT,
            user_agent TEXT
        );

        CREATE INDEX IF NOT EXISTS idx_download_logs_user_time ON download_logs(user_id, downloaded_at);
        CREATE INDEX IF NOT EXISTS idx_download_logs_resource ON download_logs(resource_id);

        CREATE TABLE IF NOT EXISTS audit_logs (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            actor_type TEXT NOT NULL,
            actor_id INTEGER NOT NULL,
            action TEXT NOT NULL,
            resource_type TEXT NOT NULL,
            resource_id TEXT NOT NULL,
            before_json TEXT,
            after_json TEXT,
            ip TEXT,
            created_at TEXT NOT NULL
        );

        CREATE TABLE IF NOT EXISTS user_messages (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            user_id INTEGER NOT NULL,
            title TEXT NOT NULL,
            content TEXT NOT NULL,
            created_at TEXT NOT NULL,
            read_at TEXT,
            sender_type TEXT NOT NULL DEFAULT 'SYSTEM',
            sender_id INTEGER
        );

        CREATE INDEX IF NOT EXISTS idx_user_messages_user_time ON user_messages(user_id, created_at);
        CREATE INDEX IF NOT EXISTS idx_user_messages_user_read ON user_messages(user_id, read_at);

        CREATE TABLE IF NOT EXISTS app_settings (
            key TEXT PRIMARY KEY,
            category TEXT NOT NULL DEFAULT 'GENERAL',
            value TEXT NOT NULL,
            updated_at TEXT NOT NULL
        );
        """
    )
    cols = [r["name"] for r in fetch_all("PRAGMA table_info(app_settings)", ())]
    if "category" not in cols:
        db.execute("ALTER TABLE app_settings ADD COLUMN category TEXT NOT NULL DEFAULT 'GENERAL'")
        db.execute("UPDATE app_settings SET category = 'GOGS' WHERE key LIKE 'GOGS_%'")
        db.execute("UPDATE app_settings SET category = 'PAYMENT' WHERE key LIKE 'PAY_%'")
        db.execute("UPDATE app_settings SET category = 'LLM' WHERE key LIKE 'LLM_%'")

    res_cols = [r["name"] for r in fetch_all("PRAGMA table_info(resources)", ())]
    if "repo_private" not in res_cols:
        db.execute("ALTER TABLE resources ADD COLUMN repo_private INTEGER NOT NULL DEFAULT 0")
    if "repo_html_url" not in res_cols:
        db.execute("ALTER TABLE resources ADD COLUMN repo_html_url TEXT")

    msg_cols = [r["name"] for r in fetch_all("PRAGMA table_info(user_messages)", ())]
    if "sender_type" not in msg_cols:
        db.execute("ALTER TABLE user_messages ADD COLUMN sender_type TEXT NOT NULL DEFAULT 'SYSTEM'")
    if "sender_id" not in msg_cols:
        db.execute("ALTER TABLE user_messages ADD COLUMN sender_id INTEGER")
    db.commit()


def ensure_default_admin() -> None:
    existing = fetch_one("SELECT id FROM admin_users WHERE username = ?", ("admin",))
    if existing is not None:
        return
    default_password = os.environ.get("ADMIN_INIT_PASSWORD", "admin123")
    execute(
        "INSERT INTO admin_users (username, password_hash, role, status) VALUES (?, ?, ?, ?)",
        ("admin", generate_password_hash(default_password), "ADMIN", "ACTIVE"),
    )


def ensure_default_plans() -> None:
    row = fetch_one("SELECT COUNT(1) AS cnt FROM plans", ())
    if row is not None and int(row["cnt"]) > 0:
        return
    defaults = [
        ("月卡", 30, 1999, 1, 30),
        ("季卡", 90, 4999, 1, 20),
        ("年卡", 365, 14999, 1, 10),
    ]
    for name, days, price_cents, enabled, sort in defaults:
        execute(
            "INSERT INTO plans (name, duration_days, price_cents, enabled, sort) VALUES (?, ?, ?, ?, ?)",
            (name, days, price_cents, enabled, sort),
        )


def db_status() -> dict[str, Any]:
    config = get_config()
    backend = _resolve_backend_from_control()
    if backend is None:
        backend = "mysql" if (config.mysql_host or "").strip() else "sqlite"
    return {
        "active": backend,
        "migrating": is_migrating(),
        "mysql": {
            "host": (_ctl_get_value("MYSQL_HOST") or config.mysql_host or "").strip(),
            "port": int((_ctl_get_value("MYSQL_PORT") or str(config.mysql_port or 3306)).strip() or 3306),
            "user": (_ctl_get_value("MYSQL_USER") or config.mysql_user or "").strip(),
            "database": (_ctl_get_value("MYSQL_DATABASE") or config.mysql_database or "").strip(),
            "hasPassword": bool((_ctl_get_value("MYSQL_PASSWORD") or config.mysql_password or "").strip()),
        },
    }


def switch_database(*, target: str, force: bool) -> dict[str, Any]:
    t = (target or "").strip().lower()
    if t not in {"sqlite", "mysql"}:
        raise RuntimeError("invalid_target")
    with _db_switch_lock:
        if is_migrating():
            raise RuntimeError("migration_running")
        config = get_config()
        cur_backend = _resolve_backend_from_control()
        if cur_backend is None:
            cur_backend = "mysql" if (config.mysql_host or "").strip() else "sqlite"
        if cur_backend == t:
            set_active_backend(cur_backend)
            return {"ok": True, "from": cur_backend, "to": t, "noop": True}

        mysql_host = (_ctl_get_value("MYSQL_HOST") or config.mysql_host or "").strip()
        mysql_user = (_ctl_get_value("MYSQL_USER") or config.mysql_user or "").strip()
        mysql_database = (_ctl_get_value("MYSQL_DATABASE") or config.mysql_database or "").strip()
        mysql_port = int((_ctl_get_value("MYSQL_PORT") or str(config.mysql_port or 3306)).strip() or 3306)
        mysql_password = (_ctl_get_value("MYSQL_PASSWORD") or config.mysql_password or "").strip()

        tables: list[tuple[str, list[str]]] = [
            ("users", ["id", "phone", "password_hash", "status", "vip_expire_at", "created_at"]),
            ("admin_users", ["id", "username", "password_hash", "role", "status", "last_login_at"]),
            ("plans", ["id", "name", "duration_days", "price_cents", "enabled", "sort"]),
            (
                "resources",
                [
                    "id",
                    "title",
                    "summary",
                    "type",
                    "status",
                    "cover_url",
                    "category",
                    "tags_json",
                    "repo_owner",
                    "repo_name",
                    "repo_private",
                    "repo_html_url",
                    "default_ref",
                    "view_count",
                    "download_count",
                    "created_at",
                    "updated_at",
                ],
            ),
            (
                "orders",
                [
                    "id",
                    "user_id",
                    "plan_id",
                    "amount_cents",
                    "status",
                    "pay_channel",
                    "pay_trade_no",
                    "created_at",
                    "paid_at",
                    "plan_snapshot_json",
                ],
            ),
            (
                "download_logs",
                [
                    "id",
                    "user_id",
                    "resource_id",
                    "resource_title_snapshot",
                    "resource_type_snapshot",
                    "ref_snapshot",
                    "downloaded_at",
                    "ip",
                    "user_agent",
                ],
            ),
            (
                "audit_logs",
                [
                    "id",
                    "actor_type",
                    "actor_id",
                    "action",
                    "resource_type",
                    "resource_id",
                    "before_json",
                    "after_json",
                    "ip",
                    "created_at",
                ],
            ),
            (
                "user_messages",
                ["id", "user_id", "title", "content", "created_at", "read_at", "sender_type", "sender_id"],
            ),
        ]

        def _mysql_connect() -> Any:
            if pymysql is None:
                raise RuntimeError("pymysql_required")
            try:
                return pymysql.connect(
                    host=mysql_host,
                    port=mysql_port,
                    user=mysql_user,
                    password=mysql_password,
                    database=mysql_database,
                    charset="utf8mb4",
                    cursorclass=pymysql.cursors.DictCursor,
                    autocommit=False,
                )
            except Exception as e:
                errno = int(getattr(e, "args", [0])[0] or 0) if getattr(e, "args", None) else 0
                if errno != 1049:
                    raise RuntimeError("connect_failed")
                server_conn = None
                try:
                    server_conn = pymysql.connect(
                        host=mysql_host,
                        port=mysql_port,
                        user=mysql_user,
                        password=mysql_password,
                        charset="utf8mb4",
                        cursorclass=pymysql.cursors.DictCursor,
                        autocommit=True,
                    )
                    esc = mysql_database.replace("`", "``")
                    cur = server_conn.cursor()
                    cur.execute(
                        f"CREATE DATABASE IF NOT EXISTS `{esc}` CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci"
                    )
                except Exception:
                    raise RuntimeError("db_create_failed")
                finally:
                    if server_conn is not None:
                        try:
                            server_conn.close()
                        except Exception:
                            pass
                try:
                    return pymysql.connect(
                        host=mysql_host,
                        port=mysql_port,
                        user=mysql_user,
                        password=mysql_password,
                        database=mysql_database,
                        charset="utf8mb4",
                        cursorclass=pymysql.cursors.DictCursor,
                        autocommit=False,
                    )
                except Exception:
                    raise RuntimeError("connect_failed")

        def _count(conn: Any, backend: str, table: str) -> int:
            if backend == "sqlite":
                row = conn.execute(f"SELECT COUNT(1) AS cnt FROM {table}").fetchone()
                return int(row["cnt"] if row is not None else 0)
            cur = conn.cursor()
            cur.execute(f"SELECT COUNT(1) AS cnt FROM {table}")
            row = cur.fetchone()
            return int((row or {}).get("cnt") or 0)

        def _wipe(conn: Any, backend: str) -> None:
            if backend == "sqlite":
                for table, _cols in tables:
                    conn.execute(f"DELETE FROM {table}")
                try:
                    conn.execute("DELETE FROM sqlite_sequence")
                except Exception:
                    pass
                return
            cur = conn.cursor()
            cur.execute("SET FOREIGN_KEY_CHECKS=0")
            for table, _cols in tables:
                cur.execute(f"TRUNCATE TABLE {table}")
            cur.execute("SET FOREIGN_KEY_CHECKS=1")

        def _insert_many(conn: Any, backend: str, table: str, cols: list[str], rows: list[list[Any]]) -> None:
            if not rows:
                return
            cols_sql = ", ".join(cols)
            if backend == "sqlite":
                q = ", ".join(["?"] * len(cols))
                conn.executemany(f"INSERT INTO {table} ({cols_sql}) VALUES ({q})", rows)
                return
            q = ", ".join(["%s"] * len(cols))
            cur = conn.cursor()
            cur.executemany(f"INSERT INTO {table} ({cols_sql}) VALUES ({q})", rows)

        def _copy(src: Any, src_backend: str, dst: Any, dst_backend: str, table: str, cols: list[str]) -> int:
            col_sql = ", ".join(cols)
            if src_backend == "sqlite":
                cur = src.execute(f"SELECT {col_sql} FROM {table}")
                src_rows = cur.fetchall() or []
                rows = [[r[c] for c in cols] for r in src_rows]
            else:
                cur = src.cursor()
                cur.execute(f"SELECT {col_sql} FROM {table}")
                src_rows = cur.fetchall() or []
                rows = [[r.get(c) for c in cols] for r in src_rows]
            _insert_many(dst, dst_backend, table, cols, rows)
            return len(rows)

        if t == "mysql" and (not mysql_host or not mysql_user or not mysql_database):
            raise RuntimeError("mysql_not_configured")

        src_backend = cur_backend
        dst_backend = t
        src_conn = None
        dst_conn = None
        try:
            _set_migrating(True)
            _ctl_set_value("DB_MIGRATING", "1", category="DB")

            src_conn = _sqlite_connect(str(config.database_path)) if src_backend == "sqlite" else _mysql_connect()
            dst_conn = _sqlite_connect(str(config.database_path)) if dst_backend == "sqlite" else _mysql_connect()

            if dst_backend == "mysql":
                stmts = [
                    """
                    CREATE TABLE IF NOT EXISTS users (
                      id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
                      phone VARCHAR(32) NOT NULL,
                      password_hash VARCHAR(255) NOT NULL,
                      status VARCHAR(16) NOT NULL DEFAULT 'ACTIVE',
                      vip_expire_at VARCHAR(64) NULL,
                      created_at VARCHAR(64) NOT NULL,
                      UNIQUE KEY uk_users_phone (phone),
                      KEY idx_users_status (status),
                      KEY idx_users_vip_expire_at (vip_expire_at)
                    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
                    """,
                    """
                    CREATE TABLE IF NOT EXISTS admin_users (
                      id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
                      username VARCHAR(64) NOT NULL,
                      password_hash VARCHAR(255) NOT NULL,
                      role VARCHAR(16) NOT NULL DEFAULT 'ADMIN',
                      status VARCHAR(16) NOT NULL DEFAULT 'ACTIVE',
                      last_login_at VARCHAR(64) NULL,
                      UNIQUE KEY uk_admin_users_username (username),
                      KEY idx_admin_users_status (status)
                    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
                    """,
                    """
                    CREATE TABLE IF NOT EXISTS plans (
                      id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
                      name VARCHAR(120) NOT NULL,
                      duration_days INT NOT NULL,
                      price_cents INT NOT NULL,
                      enabled TINYINT(1) NOT NULL DEFAULT 1,
                      sort INT NOT NULL DEFAULT 0,
                      KEY idx_plans_enabled_sort (enabled, sort)
                    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
                    """,
                    """
                    CREATE TABLE IF NOT EXISTS resources (
                      id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
                      title VARCHAR(200) NOT NULL,
                      summary MEDIUMTEXT NOT NULL,
                      type VARCHAR(8) NOT NULL,
                      status VARCHAR(16) NOT NULL DEFAULT 'DRAFT',
                      cover_url VARCHAR(500) NULL,
                      category VARCHAR(120) NULL,
                      tags_json TEXT NOT NULL,
                      repo_owner VARCHAR(120) NOT NULL,
                      repo_name VARCHAR(120) NOT NULL,
                      repo_private TINYINT(1) NOT NULL DEFAULT 0,
                      repo_html_url VARCHAR(500) NULL,
                      default_ref VARCHAR(120) NOT NULL DEFAULT 'master',
                      view_count INT NOT NULL DEFAULT 0,
                      download_count INT NOT NULL DEFAULT 0,
                      created_at VARCHAR(64) NOT NULL,
                      updated_at VARCHAR(64) NOT NULL,
                      KEY idx_resources_status_updated (status, updated_at),
                      KEY idx_resources_status_views (status, view_count),
                      KEY idx_resources_type_status (type, status)
                    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
                    """,
                    """
                    CREATE TABLE IF NOT EXISTS orders (
                      id VARCHAR(64) NOT NULL PRIMARY KEY,
                      user_id BIGINT UNSIGNED NOT NULL,
                      plan_id BIGINT UNSIGNED NOT NULL,
                      amount_cents INT NOT NULL,
                      status VARCHAR(16) NOT NULL DEFAULT 'PENDING',
                      pay_channel VARCHAR(32) NULL,
                      pay_trade_no VARCHAR(128) NULL,
                      created_at VARCHAR(64) NOT NULL,
                      paid_at VARCHAR(64) NULL,
                      plan_snapshot_json TEXT NOT NULL,
                      KEY idx_orders_user_time (user_id, created_at),
                      KEY idx_orders_status_time (status, created_at),
                      KEY idx_orders_pay_trade_no (pay_trade_no)
                    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
                    """,
                    """
                    CREATE TABLE IF NOT EXISTS download_logs (
                      id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
                      user_id BIGINT UNSIGNED NOT NULL,
                      resource_id BIGINT UNSIGNED NULL,
                      resource_title_snapshot VARCHAR(200) NOT NULL,
                      resource_type_snapshot VARCHAR(16) NOT NULL,
                      ref_snapshot VARCHAR(200) NOT NULL,
                      downloaded_at VARCHAR(64) NOT NULL,
                      ip VARCHAR(64) NULL,
                      user_agent VARCHAR(256) NULL,
                      KEY idx_download_logs_user_time (user_id, downloaded_at),
                      KEY idx_download_logs_resource (resource_id)
                    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
                    """,
                    """
                    CREATE TABLE IF NOT EXISTS audit_logs (
                      id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
                      actor_type VARCHAR(16) NOT NULL,
                      actor_id BIGINT UNSIGNED NOT NULL,
                      action VARCHAR(64) NOT NULL,
                      resource_type VARCHAR(64) NOT NULL,
                      resource_id VARCHAR(128) NOT NULL,
                      before_json MEDIUMTEXT NULL,
                      after_json MEDIUMTEXT NULL,
                      ip VARCHAR(64) NULL,
                      created_at VARCHAR(64) NOT NULL,
                      KEY idx_audit_logs_time (created_at),
                      KEY idx_audit_logs_actor (actor_type, actor_id)
                    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
                    """,
                    """
                    CREATE TABLE IF NOT EXISTS user_messages (
                      id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
                      user_id BIGINT UNSIGNED NOT NULL,
                      title VARCHAR(120) NOT NULL,
                      content MEDIUMTEXT NOT NULL,
                      created_at VARCHAR(64) NOT NULL,
                      read_at VARCHAR(64) NULL,
                      sender_type VARCHAR(16) NOT NULL DEFAULT 'SYSTEM',
                      sender_id BIGINT UNSIGNED NULL,
                      KEY idx_user_messages_user_time (user_id, created_at),
                      KEY idx_user_messages_user_read (user_id, read_at),
                      KEY idx_user_messages_sender (sender_type, created_at)
                    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
                    """,
                ]
                cur = dst_conn.cursor()
                for s in stmts:
                    cur.execute(s)
            else:
                dst_conn.executescript(
                    """
                    CREATE TABLE IF NOT EXISTS users (
                        id INTEGER PRIMARY KEY AUTOINCREMENT,
                        phone TEXT NOT NULL UNIQUE,
                        password_hash TEXT NOT NULL,
                        status TEXT NOT NULL DEFAULT 'ACTIVE',
                        vip_expire_at TEXT,
                        created_at TEXT NOT NULL
                    );
                    CREATE INDEX IF NOT EXISTS idx_users_status ON users(status);
                    CREATE INDEX IF NOT EXISTS idx_users_vip_expire_at ON users(vip_expire_at);

                    CREATE TABLE IF NOT EXISTS admin_users (
                        id INTEGER PRIMARY KEY AUTOINCREMENT,
                        username TEXT NOT NULL UNIQUE,
                        password_hash TEXT NOT NULL,
                        role TEXT NOT NULL DEFAULT 'ADMIN',
                        status TEXT NOT NULL DEFAULT 'ACTIVE',
                        last_login_at TEXT
                    );
                    CREATE INDEX IF NOT EXISTS idx_admin_users_status ON admin_users(status);

                    CREATE TABLE IF NOT EXISTS plans (
                        id INTEGER PRIMARY KEY AUTOINCREMENT,
                        name TEXT NOT NULL,
                        duration_days INTEGER NOT NULL,
                        price_cents INTEGER NOT NULL,
                        enabled INTEGER NOT NULL DEFAULT 1,
                        sort INTEGER NOT NULL DEFAULT 0
                    );

                    CREATE TABLE IF NOT EXISTS resources (
                        id INTEGER PRIMARY KEY AUTOINCREMENT,
                        title TEXT NOT NULL,
                        summary TEXT NOT NULL DEFAULT '',
                        type TEXT NOT NULL CHECK (type IN ('FREE', 'VIP')),
                        status TEXT NOT NULL CHECK (status IN ('DRAFT', 'ONLINE', 'OFFLINE')) DEFAULT 'DRAFT',
                        cover_url TEXT,
                        category TEXT,
                        tags_json TEXT NOT NULL DEFAULT '[]',
                        repo_owner TEXT NOT NULL,
                        repo_name TEXT NOT NULL,
                        repo_private INTEGER NOT NULL DEFAULT 0,
                        repo_html_url TEXT,
                        default_ref TEXT NOT NULL DEFAULT 'master',
                        view_count INTEGER NOT NULL DEFAULT 0,
                        download_count INTEGER NOT NULL DEFAULT 0,
                        created_at TEXT NOT NULL,
                        updated_at TEXT NOT NULL
                    );
                    CREATE INDEX IF NOT EXISTS idx_resources_status_updated ON resources(status, updated_at);
                    CREATE INDEX IF NOT EXISTS idx_resources_status_views ON resources(status, view_count);
                    CREATE INDEX IF NOT EXISTS idx_resources_type_status ON resources(type, status);

                    CREATE TABLE IF NOT EXISTS orders (
                        id TEXT PRIMARY KEY,
                        user_id INTEGER NOT NULL,
                        plan_id INTEGER NOT NULL,
                        amount_cents INTEGER NOT NULL,
                        status TEXT NOT NULL CHECK (status IN ('PENDING', 'PAID', 'CLOSED', 'FAILED')) DEFAULT 'PENDING',
                        pay_channel TEXT,
                        pay_trade_no TEXT,
                        created_at TEXT NOT NULL,
                        paid_at TEXT,
                        plan_snapshot_json TEXT NOT NULL
                    );
                    CREATE INDEX IF NOT EXISTS idx_orders_user_time ON orders(user_id, created_at);
                    CREATE INDEX IF NOT EXISTS idx_orders_status_time ON orders(status, created_at);
                    CREATE INDEX IF NOT EXISTS idx_orders_pay_trade_no ON orders(pay_trade_no);

                    CREATE TABLE IF NOT EXISTS download_logs (
                        id INTEGER PRIMARY KEY AUTOINCREMENT,
                        user_id INTEGER NOT NULL,
                        resource_id INTEGER,
                        resource_title_snapshot TEXT NOT NULL,
                        resource_type_snapshot TEXT NOT NULL,
                        ref_snapshot TEXT NOT NULL,
                        downloaded_at TEXT NOT NULL,
                        ip TEXT,
                        user_agent TEXT
                    );
                    CREATE INDEX IF NOT EXISTS idx_download_logs_user_time ON download_logs(user_id, downloaded_at);
                    CREATE INDEX IF NOT EXISTS idx_download_logs_resource ON download_logs(resource_id);

                    CREATE TABLE IF NOT EXISTS audit_logs (
                        id INTEGER PRIMARY KEY AUTOINCREMENT,
                        actor_type TEXT NOT NULL,
                        actor_id INTEGER NOT NULL,
                        action TEXT NOT NULL,
                        resource_type TEXT NOT NULL,
                        resource_id TEXT NOT NULL,
                        before_json TEXT,
                        after_json TEXT,
                        ip TEXT,
                        created_at TEXT NOT NULL
                    );

                    CREATE TABLE IF NOT EXISTS user_messages (
                        id INTEGER PRIMARY KEY AUTOINCREMENT,
                        user_id INTEGER NOT NULL,
                        title TEXT NOT NULL,
                        content TEXT NOT NULL,
                        created_at TEXT NOT NULL,
                        read_at TEXT,
                        sender_type TEXT NOT NULL DEFAULT 'SYSTEM',
                        sender_id INTEGER
                    );
                    CREATE INDEX IF NOT EXISTS idx_user_messages_user_time ON user_messages(user_id, created_at);
                    CREATE INDEX IF NOT EXISTS idx_user_messages_user_read ON user_messages(user_id, read_at);
                    """
                )

            target_has = {table: _count(dst_conn, dst_backend, table) for table, _cols in tables}
            if not force and any(v > 0 for v in target_has.values()):
                raise RuntimeError("target_not_empty")

            _wipe(dst_conn, dst_backend)

            migrated: dict[str, int] = {}
            for table, cols in tables:
                migrated[table] = _copy(src_conn, src_backend, dst_conn, dst_backend, table, cols)

            for table, _cols in tables:
                if _count(src_conn, src_backend, table) != _count(dst_conn, dst_backend, table):
                    raise RuntimeError("verify_failed")

            dst_conn.commit()
            _ctl_set_value("DB_ACTIVE", dst_backend, category="DB")
            set_active_backend(dst_backend)
            return {"ok": True, "from": src_backend, "to": dst_backend, "migrated": migrated}
        except Exception:
            if dst_conn is not None:
                try:
                    dst_conn.rollback()
                except Exception:
                    pass
            raise
        finally:
            _ctl_set_value("DB_MIGRATING", "0", category="DB")
            _set_migrating(False)
            if src_conn is not None:
                try:
                    src_conn.close()
                except Exception:
                    pass
            if dst_conn is not None:
                try:
                    dst_conn.close()
                except Exception:
                    pass