"""兑换码的 SQLite 存储与原子领取逻辑。"""

from __future__ import annotations

import asyncio
import sqlite3
import threading
from contextlib import contextmanager
from datetime import UTC, datetime, timedelta, timezone
from pathlib import Path
from typing import Any


BEIJING_TZ = timezone(timedelta(hours=8), name="Asia/Shanghai")


def normalize_code(value: str) -> str:
    return str(value or "").strip().casefold()


def utc_now() -> str:
    return datetime.now(UTC).isoformat(timespec="milliseconds").replace("+00:00", "Z")


def parse_beijing_datetime(value: str) -> str:
    raw = str(value or "").strip()
    if not raw:
        raise ValueError("生效时间和失效时间不能为空")
    try:
        parsed = datetime.fromisoformat(raw.replace("Z", "+00:00"))
    except ValueError as exc:
        raise ValueError("时间格式无效") from exc
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=BEIJING_TZ)
    return parsed.astimezone(UTC).isoformat(timespec="milliseconds").replace("+00:00", "Z")


def to_beijing_datetime(value: str | None) -> str:
    if not value:
        return ""
    parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=UTC)
    return parsed.astimezone(BEIJING_TZ).isoformat(timespec="minutes")


_SCHEMA = """
CREATE TABLE IF NOT EXISTS redemption_codes (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    code_normalized TEXT NOT NULL UNIQUE,
    code_display TEXT NOT NULL,
    points INTEGER NOT NULL CHECK(points > 0),
    starts_at_utc TEXT NOT NULL,
    ends_at_utc TEXT NOT NULL,
    max_redemptions INTEGER CHECK(max_redemptions IS NULL OR max_redemptions > 0),
    redeemed_count INTEGER NOT NULL DEFAULT 0 CHECK(redeemed_count >= 0),
    enabled INTEGER NOT NULL DEFAULT 1 CHECK(enabled IN (0, 1)),
    deleted_at_utc TEXT,
    created_at_utc TEXT NOT NULL,
    updated_at_utc TEXT NOT NULL,
    CHECK(starts_at_utc < ends_at_utc)
);
CREATE TABLE IF NOT EXISTS redemption_records (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    code_id INTEGER NOT NULL REFERENCES redemption_codes(id),
    code_display TEXT NOT NULL,
    platform TEXT NOT NULL,
    platform_user_id TEXT NOT NULL,
    display_name TEXT NOT NULL,
    points INTEGER NOT NULL,
    balance_before INTEGER NOT NULL,
    balance_after INTEGER,
    status TEXT NOT NULL DEFAULT 'pending',
    redeemed_at_utc TEXT NOT NULL,
    completed_at_utc TEXT,
    UNIQUE(code_id, platform, platform_user_id)
);
CREATE INDEX IF NOT EXISTS idx_redemption_codes_active
    ON redemption_codes(deleted_at_utc, enabled, starts_at_utc, ends_at_utc);
CREATE INDEX IF NOT EXISTS idx_redemption_records_code_time
    ON redemption_records(code_id, redeemed_at_utc DESC);
CREATE INDEX IF NOT EXISTS idx_redemption_records_user_time
    ON redemption_records(platform, platform_user_id, redeemed_at_utc DESC);
"""


class RedemptionCodeStore:
    """用短事务处理兑换码，避免并发超领和重复领取。

    当提供 stats_store 时，所有读写都复用 stats_store 的唯一连接与串行写队列，
    消除多连接写同一文件导致的 "database is locked" 争用。
    """

    def __init__(self, database_path: str | Path, stats_store: Any | None = None):
        self.database_path = Path(database_path)
        self._store = stats_store
        self._lock = threading.RLock()
        self.database_path.parent.mkdir(parents=True, exist_ok=True)
        with self._connect() as connection:
            connection.executescript(_SCHEMA)
            connection.commit()

    @contextmanager
    def _connect(self):
        connection = sqlite3.connect(self.database_path, timeout=5.0)
        connection.row_factory = sqlite3.Row
        connection.execute("PRAGMA journal_mode=WAL")
        connection.execute("PRAGMA synchronous=NORMAL")
        connection.execute("PRAGMA foreign_keys=ON")
        connection.execute("PRAGMA busy_timeout=5000")
        try:
            yield connection
        finally:
            connection.close()

    def _run(self, func, *args):
        """在 stats_store 单连接（或自有连接）上执行 func(connection, *args)。"""
        if self._store is not None:
            return self._store.execute(func, *args)
        return asyncio.to_thread(self._run_direct, func, args)

    def _run_direct(self, func, args):
        with self._lock, self._connect() as connection:
            return func(connection, *args)

    @staticmethod
    def _serialize_code(row: sqlite3.Row) -> dict[str, Any]:
        max_redemptions = row["max_redemptions"]
        redeemed_count = int(row["redeemed_count"] or 0)
        return {
            "id": int(row["id"]),
            "code": row["code_display"],
            "points": int(row["points"]),
            "starts_at": to_beijing_datetime(row["starts_at_utc"]),
            "ends_at": to_beijing_datetime(row["ends_at_utc"]),
            "max_redemptions": int(max_redemptions) if max_redemptions is not None else None,
            "redeemed_count": redeemed_count,
            "remaining_count": max(0, int(max_redemptions) - redeemed_count) if max_redemptions is not None else None,
            "enabled": bool(row["enabled"]),
            "created_at": to_beijing_datetime(row["created_at_utc"]),
            "updated_at": to_beijing_datetime(row["updated_at_utc"]),
        }

    @staticmethod
    def _serialize_record(row: sqlite3.Row) -> dict[str, Any]:
        return {
            "id": int(row["id"]),
            "code_id": int(row["code_id"]),
            "code": row["code_display"],
            "platform": row["platform"],
            "uid": row["platform_user_id"],
            "uname": row["display_name"],
            "points": int(row["points"]),
            "balance_before": int(row["balance_before"]),
            "balance_after": int(row["balance_after"]) if row["balance_after"] is not None else None,
            "status": row["status"],
            "redeemed_at": to_beijing_datetime(row["redeemed_at_utc"]),
            "completed_at": to_beijing_datetime(row["completed_at_utc"]),
        }

    async def create_code(
        self,
        *,
        code: str,
        points: int,
        starts_at: str,
        ends_at: str,
        max_redemptions: int | None,
        enabled: bool = True,
    ) -> dict[str, Any]:
        return await self._run(
            self._create_code,
            code,
            points,
            starts_at,
            ends_at,
            max_redemptions,
            enabled,
        )

    def _create_code(
        self,
        connection,
        code: str,
        points: int,
        starts_at: str,
        ends_at: str,
        max_redemptions: int | None,
        enabled: bool,
    ) -> dict[str, Any]:
        display = str(code or "").strip()
        normalized = normalize_code(display)
        if not normalized:
            raise ValueError("兑换码不能为空")
        if len(display) > 100:
            raise ValueError("兑换码不能超过100个字符")
        points = int(points)
        if points <= 0:
            raise ValueError("兑换积分必须是正整数")
        if max_redemptions in ("", None):
            max_value = None
        else:
            max_value = int(max_redemptions)
            if max_value <= 0:
                raise ValueError("总兑换次数必须是正整数，或留空表示不限")
        starts_utc = parse_beijing_datetime(starts_at)
        ends_utc = parse_beijing_datetime(ends_at)
        if starts_utc >= ends_utc:
            raise ValueError("失效时间必须晚于生效时间")
        now = utc_now()
        try:
            cursor = connection.execute(
                """INSERT INTO redemption_codes (
                    code_normalized, code_display, points, starts_at_utc, ends_at_utc,
                    max_redemptions, enabled, created_at_utc, updated_at_utc
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
                (normalized, display, points, starts_utc, ends_utc, max_value, int(bool(enabled)), now, now),
            )
            connection.commit()
        except sqlite3.IntegrityError as exc:
            connection.rollback()
            if "code_normalized" in str(exc) or "UNIQUE constraint" in str(exc):
                raise ValueError("兑换码已存在") from exc
            raise
        row = connection.execute("SELECT * FROM redemption_codes WHERE id=?", (cursor.lastrowid,)).fetchone()
        assert row is not None
        return self._serialize_code(row)

    async def list_codes(self) -> list[dict[str, Any]]:
        return await self._run(self._list_codes)

    def _list_codes(self, connection) -> list[dict[str, Any]]:
        rows = connection.execute(
            "SELECT * FROM redemption_codes WHERE deleted_at_utc IS NULL ORDER BY id DESC"
        ).fetchall()
        return [self._serialize_code(row) for row in rows]

    async def list_records(self, *, code_id: int | None = None, limit: int = 500) -> list[dict[str, Any]]:
        return await self._run(self._list_records, code_id, limit)

    def _list_records(self, connection, code_id: int | None, limit: int) -> list[dict[str, Any]]:
        limit = max(1, min(2000, int(limit)))
        if code_id is None:
            rows = connection.execute(
                "SELECT * FROM redemption_records ORDER BY id DESC LIMIT ?", (limit,)
            ).fetchall()
        else:
            rows = connection.execute(
                "SELECT * FROM redemption_records WHERE code_id=? ORDER BY id DESC LIMIT ?",
                (int(code_id), limit),
            ).fetchall()
        return [self._serialize_record(row) for row in rows]

    async def set_enabled(self, code_id: int, enabled: bool) -> dict[str, Any]:
        return await self._run(self._set_enabled, code_id, enabled)

    def _set_enabled(self, connection, code_id: int, enabled: bool) -> dict[str, Any]:
        cursor = connection.execute(
            "UPDATE redemption_codes SET enabled=?, updated_at_utc=? WHERE id=? AND deleted_at_utc IS NULL",
            (int(bool(enabled)), utc_now(), int(code_id)),
        )
        if cursor.rowcount != 1:
            connection.rollback()
            raise ValueError("兑换码不存在")
        connection.commit()
        row = connection.execute("SELECT * FROM redemption_codes WHERE id=?", (int(code_id),)).fetchone()
        assert row is not None
        return self._serialize_code(row)

    async def delete_code(self, code_id: int) -> None:
        await self._run(self._delete_code, code_id)

    def _delete_code(self, connection, code_id: int) -> None:
        now = utc_now()
        cursor = connection.execute(
            """UPDATE redemption_codes
               SET enabled=0, deleted_at_utc=?, updated_at_utc=?
               WHERE id=? AND deleted_at_utc IS NULL""",
            (now, now, int(code_id)),
        )
        if cursor.rowcount != 1:
            connection.rollback()
            raise ValueError("兑换码不存在")
        connection.commit()

    async def reserve(
        self,
        message: str,
        *,
        platform: str,
        platform_user_id: str,
        display_name: str,
        balance_before: int,
    ) -> dict[str, Any]:
        return await self._run(
            self._reserve,
            message,
            platform,
            platform_user_id,
            display_name,
            balance_before,
        )

    def _reserve(
        self,
        connection,
        message: str,
        platform: str,
        platform_user_id: str,
        display_name: str,
        balance_before: int,
    ) -> dict[str, Any]:
        normalized = normalize_code(message)
        if not normalized:
            return {"status": "unknown"}
        now = utc_now()
        connection.execute("BEGIN IMMEDIATE")
        try:
            code = connection.execute(
                "SELECT * FROM redemption_codes WHERE code_normalized=? AND deleted_at_utc IS NULL",
                (normalized,),
            ).fetchone()
            if code is None:
                connection.rollback()
                return {"status": "unknown"}
            if not bool(code["enabled"]):
                connection.rollback()
                return {"status": "disabled"}
            if now < code["starts_at_utc"]:
                connection.rollback()
                return {"status": "not_started"}
            if now >= code["ends_at_utc"]:
                connection.rollback()
                return {"status": "expired"}
            existing = connection.execute(
                """SELECT status FROM redemption_records
                   WHERE code_id=? AND platform=? AND platform_user_id=?""",
                (code["id"], platform, platform_user_id),
            ).fetchone()
            if existing is not None:
                connection.rollback()
                return {"status": "already_redeemed"}
            max_redemptions = code["max_redemptions"]
            if max_redemptions is not None and int(code["redeemed_count"]) >= int(max_redemptions):
                connection.rollback()
                return {"status": "exhausted"}
            cursor = connection.execute(
                """INSERT INTO redemption_records (
                    code_id, code_display, platform, platform_user_id, display_name,
                    points, balance_before, status, redeemed_at_utc
                ) VALUES (?, ?, ?, ?, ?, ?, ?, 'pending', ?)""",
                (
                    code["id"], code["code_display"], platform, platform_user_id,
                    display_name, code["points"], int(balance_before), now,
                ),
            )
            updated = connection.execute(
                """UPDATE redemption_codes
                   SET redeemed_count=redeemed_count+1, updated_at_utc=?
                   WHERE id=? AND (max_redemptions IS NULL OR redeemed_count < max_redemptions)""",
                (now, code["id"]),
            )
            if updated.rowcount != 1:
                connection.rollback()
                return {"status": "exhausted"}
            connection.commit()
            return {
                "status": "reserved",
                "record_id": int(cursor.lastrowid),
                "code_id": int(code["id"]),
                "code": code["code_display"],
                "points": int(code["points"]),
            }
        except sqlite3.IntegrityError:
            connection.rollback()
            return {"status": "already_redeemed"}
        except Exception:
            connection.rollback()
            raise

    async def finalize(self, record_id: int, balance_after: int) -> None:
        await self._run(self._finalize, record_id, balance_after)

    def _finalize(self, connection, record_id: int, balance_after: int) -> None:
        cursor = connection.execute(
            """UPDATE redemption_records
               SET status='completed', balance_after=?, completed_at_utc=?
               WHERE id=? AND status='pending'""",
            (int(balance_after), utc_now(), int(record_id)),
        )
        if cursor.rowcount != 1:
            connection.rollback()
            raise ValueError("兑换记录不存在或已经完成")
        connection.commit()

    async def cancel(self, record_id: int) -> None:
        await self._run(self._cancel, record_id)

    def _cancel(self, connection, record_id: int) -> None:
        connection.execute("BEGIN IMMEDIATE")
        try:
            row = connection.execute(
                "SELECT code_id FROM redemption_records WHERE id=? AND status='pending'",
                (int(record_id),),
            ).fetchone()
            if row is None:
                connection.rollback()
                return
            connection.execute("DELETE FROM redemption_records WHERE id=?", (int(record_id),))
            connection.execute(
                """UPDATE redemption_codes
                   SET redeemed_count=MAX(0, redeemed_count-1), updated_at_utc=?
                   WHERE id=?""",
                (utc_now(), int(row["code_id"])),
            )
            connection.commit()
        except Exception:
            connection.rollback()
            raise