study

Paused

study / services /auth_service.py
xiaolongmr
xiufu shijian xunhuan bei suoding de genyin: qingqiu relu bu zai xie ku
0e5331b 10 days ago
142 kB
from __future__ import annotations

import hashlib
import hmac
import secrets
import time
import uuid
from datetime import datetime, timedelta, timezone
from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
from threading import RLock, Thread
from typing import Callable, Literal, TypeVar

try:
    from zoneinfo import ZoneInfo
except ImportError:  # pragma: no cover - Python 3.9+ normally has zoneinfo.
    ZoneInfo = None  # type: ignore[assignment]

from services.config import config
from services.storage.base import StorageBackend

AuthRole = Literal["admin", "user", "normal"]
ResultT = TypeVar("ResultT")
THIRD_PARTY_IMAGE_GROUP_WALLET_MODEL_ID = "__group__"
DAILY_GIFT_ACCOUNT_POOL_MODEL_ID = "__account_pool__"


def _now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


def _today_key() -> str:
    tz = ZoneInfo("Asia/Shanghai") if ZoneInfo else timezone(timedelta(hours=8))
    return datetime.now(tz).strftime("%Y-%m-%d")


def _china_now() -> datetime:
    tz = ZoneInfo("Asia/Shanghai") if ZoneInfo else timezone(timedelta(hours=8))
    return datetime.now(tz)


def _period_keys(now: datetime | None = None) -> tuple[str, str, str]:
    current = now or _china_now()
    date_key = current.strftime("%Y-%m-%d")
    iso_year, iso_week, _ = current.isocalendar()
    return date_key, f"{iso_year}-W{iso_week:02d}", current.strftime("%Y-%m")


def _hash_key(value: str) -> str:
    return hashlib.sha256(value.encode("utf-8")).hexdigest()


def _invite_code() -> str:
    return secrets.token_urlsafe(6).replace("-", "").replace("_", "").upper()[:8]


def _to_nonnegative_int(value: object) -> int:
    try:
        return max(0, int(value or 0))
    except (TypeError, ValueError):
        return 0


CREDIT_QUANTUM = Decimal("0.01")


def _credit_decimal(value: object) -> Decimal:
    try:
        parsed = Decimal(str(value or 0))
    except (InvalidOperation, TypeError, ValueError):
        return Decimal("0.00")
    return max(Decimal("0"), parsed).quantize(CREDIT_QUANTUM, rounding=ROUND_HALF_UP)


def _credit_number(value: object) -> int | float:
    parsed = _credit_decimal(value)
    return int(parsed) if parsed == parsed.to_integral_value() else float(parsed)


def _parse_iso_datetime(value: object) -> datetime | None:
    raw = str(value or "").strip()
    if not raw:
        return None
    try:
        parsed = datetime.fromisoformat(raw.replace("Z", "+00:00"))
    except ValueError:
        return None
    return parsed if parsed.tzinfo is not None else parsed.replace(tzinfo=timezone.utc)


def _is_future_datetime(value: object) -> bool:
    parsed = _parse_iso_datetime(value)
    return parsed is not None and parsed > datetime.now(timezone.utc)


def _normalize_credit_grants(value: object) -> list[dict[str, object]]:
    source = value if isinstance(value, list) else []
    grants: list[dict[str, object]] = []
    for raw in source:
        if not isinstance(raw, dict):
            continue
        grant_id = str(raw.get("id") or "").strip() or uuid.uuid4().hex[:12]
        expires_at = str(raw.get("expires_at") or "").strip()
        if not expires_at:
            continue
        grants.append(
            {
                "id": grant_id,
                "remaining": _credit_number(raw.get("remaining")),
                "reserved": _credit_number(raw.get("reserved")),
                "expires_at": expires_at,
                "created_at": str(raw.get("created_at") or "").strip() or _now_iso(),
                "source": str(raw.get("source") or "").strip(),
                "code_id": str(raw.get("code_id") or "").strip(),
            }
        )
    grants.sort(key=lambda item: str(item.get("expires_at") or ""))
    return grants


def _active_credit_grants(item: dict[str, object]) -> list[dict[str, object]]:
    return [grant for grant in _normalize_credit_grants(item.get("credit_grants")) if _is_future_datetime(grant.get("expires_at"))]


def _expiring_credit_remaining(item: dict[str, object]) -> int | float:
    return _credit_number(sum((_credit_decimal(grant.get("remaining")) for grant in _active_credit_grants(item)), Decimal("0")))


def _expiring_credit_reserved(item: dict[str, object]) -> int | float:
    return _credit_number(sum((_credit_decimal(grant.get("reserved")) for grant in _normalize_credit_grants(item.get("credit_grants"))), Decimal("0")))


CHECKIN_CREDIT_SOURCE = "checkin"


def _checkin_credit_state(item: dict[str, object]) -> tuple[Decimal, str]:
    """返回 (该用户还没用掉的签到积分, 最近到期时间)。
    签到赠送积分有两种形态，取决于后台「签到积分可用天数」：
      - 永久形态（可用天数 = 0，默认）：直接进基础积分池，由 `checkin_credit_remaining` 账本跟踪；
      - 限时形态（可用天数 > 0）：发放成 `source="checkin"` 的限时券，剩余量记在券上，到期即作废。
    两者相加才是「未使用的签到积分」；`expires_at` 取未过期签到券里最早的那张（没有则为空串）。
    """
    base_balance = _credit_decimal(item.get("credits")) + _credit_decimal(item.get("reserved_credits"))
    total = min(_credit_decimal(item.get("checkin_credit_remaining")), base_balance)
    expires_at = ""
    for grant in _active_credit_grants(item):
        if str(grant.get("source") or "").strip() != CHECKIN_CREDIT_SOURCE:
            continue
        remaining = _credit_decimal(grant.get("remaining"))
        total += remaining
        # 剩余为 0 的券（已被扣光但还没清理）不该参与「最近到期时间」，否则会把它当成最近的一笔。
        if remaining <= 0:
            continue
        candidate = str(grant.get("expires_at") or "").strip()
        if candidate and (not expires_at or candidate < expires_at):
            expires_at = candidate
    return total, expires_at


def _checkin_credit_batches(item: dict[str, object]) -> list[dict[str, object]]:
    """把「未过期的签到限时券」按到期时刻聚合成 [{credits, expires_at}]，按到期时间升序。
    签到积分逐笔到期（每笔自发放起 N 天后失效），所以这里会天然出现多批；同一时刻到期的
    券（例如后台保存设置触发重算后对齐的那批）会合并成一条。
    永久形态（可用天数 = 0）没有到期时间，不产生批次 —— 前端据此隐藏到期那几行。
    """
    grouped: dict[str, Decimal] = {}
    for grant in _active_credit_grants(item):
        if str(grant.get("source") or "").strip() != CHECKIN_CREDIT_SOURCE:
            continue
        remaining = _credit_decimal(grant.get("remaining"))
        if remaining <= 0:
            continue
        key = str(grant.get("expires_at") or "").strip()
        grouped[key] = grouped.get(key, Decimal("0")) + remaining
    return [
        {"credits": _credit_number(total), "expires_at": expires_at}
        for expires_at, total in sorted(grouped.items(), key=lambda pair: pair[0])
    ]


PACKAGE_UNLIMITED_SENTINEL = 10**9


def _normalize_package_grants(value: object) -> list[dict[str, object]]:
    source = value if isinstance(value, list) else []
    grants: list[dict[str, object]] = []
    for raw in source:
        if not isinstance(raw, dict):
            continue
        grant_id = str(raw.get("id") or "").strip() or uuid.uuid4().hex[:12]
        expires_at = str(raw.get("expires_at") or "").strip()
        if not expires_at:
            continue
        grants.append(
            {
                "id": grant_id,
                "name": str(raw.get("name") or "套餐").strip() or "套餐",
                "created_at": str(raw.get("created_at") or "").strip() or _now_iso(),
                "starts_at": str(raw.get("starts_at") or "").strip(),
                "expires_at": expires_at,
                "source": str(raw.get("source") or "").strip(),
                "code_id": str(raw.get("code_id") or "").strip(),
                "daily_limit": _to_nonnegative_int(raw.get("daily_limit")),
                "weekly_limit": _to_nonnegative_int(raw.get("weekly_limit")),
                "monthly_limit": _to_nonnegative_int(raw.get("monthly_limit")),
                "total_limit": _to_nonnegative_int(raw.get("total_limit")),
                "used_total": _credit_number(raw.get("used_total")),
                "reserved_total": _credit_number(raw.get("reserved_total")),
                "daily_key": str(raw.get("daily_key") or "").strip(),
                "used_daily": _credit_number(raw.get("used_daily")),
                "reserved_daily": _credit_number(raw.get("reserved_daily")),
                "weekly_key": str(raw.get("weekly_key") or "").strip(),
                "used_weekly": _credit_number(raw.get("used_weekly")),
                "reserved_weekly": _credit_number(raw.get("reserved_weekly")),
                "monthly_key": str(raw.get("monthly_key") or "").strip(),
                "used_monthly": _credit_number(raw.get("used_monthly")),
                "reserved_monthly": _credit_number(raw.get("reserved_monthly")),
            }
        )
    grants.sort(key=lambda item: str(item.get("expires_at") or ""))
    return grants


def _normalize_string_list(value: object) -> list[str]:
    source = value if isinstance(value, list) else []
    result: list[str] = []
    seen: set[str] = set()
    for item in source:
        text = str(item or "").strip()
        if not text or text in seen:
            continue
        seen.add(text)
        result.append(text)
    return result


def _normalize_third_party_image_wallets(value: object) -> list[dict[str, object]]:
    source = value if isinstance(value, list) else []
    wallets_by_key: dict[str, dict[str, object]] = {}
    for raw in source:
        if not isinstance(raw, dict):
            continue
        group_id = str(raw.get("group_id") or "").strip()
        model_id = str(raw.get("model_id") or "").strip() or THIRD_PARTY_IMAGE_GROUP_WALLET_MODEL_ID
        if not group_id:
            continue
        wallet_id = f"{group_id}:{model_id}"
        wallet = wallets_by_key.get(wallet_id)
        if wallet is None:
            wallet = {
                "id": wallet_id,
                "group_id": group_id,
                "model_id": model_id,
                "remaining": _credit_number(raw.get("remaining")),
                "reserved": _credit_number(raw.get("reserved")),
                "used": _credit_number(raw.get("used")),
                "total_recharged": _credit_number(raw.get("total_recharged")),
                "updated_at": str(raw.get("updated_at") or "").strip(),
            }
            wallets_by_key[wallet_id] = wallet
        else:
            wallet["remaining"] = _credit_number(_credit_decimal(wallet.get("remaining")) + _credit_decimal(raw.get("remaining")))
            wallet["reserved"] = _credit_number(_credit_decimal(wallet.get("reserved")) + _credit_decimal(raw.get("reserved")))
            wallet["used"] = _credit_number(_credit_decimal(wallet.get("used")) + _credit_decimal(raw.get("used")))
            wallet["total_recharged"] = _credit_number(_credit_decimal(wallet.get("total_recharged")) + _credit_decimal(raw.get("total_recharged")))
            updated_at = str(raw.get("updated_at") or "").strip()
            if updated_at > str(wallet.get("updated_at") or ""):
                wallet["updated_at"] = updated_at
    wallets = list(wallets_by_key.values())
    wallets.sort(key=lambda item: (str(item.get("group_id") or ""), str(item.get("model_id") or "")))
    return wallets


def _normalize_image_model_daily_limits(value: object) -> list[dict[str, object]]:
    if isinstance(value, dict):
        source = [
            {"model_id": model_id, **entry} if isinstance(entry, dict) else {"model_id": model_id, "daily_limit": entry}
            for model_id, entry in value.items()
        ]
    else:
        source = value if isinstance(value, list) else []
    limits_by_model: dict[str, dict[str, object]] = {}
    for raw in source:
        if not isinstance(raw, dict):
            continue
        model_id = str(raw.get("model_id") or raw.get("model") or raw.get("id") or "").strip()
        if not model_id:
            continue
        previous = limits_by_model.get(model_id)
        used = _to_nonnegative_int(raw.get("used"))
        reserved = _to_nonnegative_int(raw.get("reserved"))
        if previous is not None and str(previous.get("date") or "") == str(raw.get("date") or raw.get("daily_key") or "").strip():
            used += _to_nonnegative_int(previous.get("used"))
            reserved += _to_nonnegative_int(previous.get("reserved"))
        limits_by_model[model_id] = {
            "model_id": model_id,
            "daily_limit": _to_nonnegative_int(raw.get("daily_limit") if "daily_limit" in raw else raw.get("limit")),
            "date": str(raw.get("date") or raw.get("daily_key") or "").strip(),
            "used": used,
            "reserved": reserved,
            "updated_at": str(raw.get("updated_at") or "").strip(),
        }
    limits = list(limits_by_model.values())
    limits.sort(key=lambda item: str(item.get("model_id") or ""))
    return limits


def _reset_image_model_daily_limit(entry: dict[str, object]) -> dict[str, object]:
    today = _today_key()
    next_entry = dict(entry)
    if str(next_entry.get("date") or "").strip() != today and _to_nonnegative_int(next_entry.get("reserved")) <= 0:
        next_entry["date"] = today
        next_entry["used"] = 0
        next_entry["reserved"] = 0
    return next_entry


def _image_model_daily_limit_public(entry: dict[str, object]) -> dict[str, object]:
    normalized = _reset_image_model_daily_limit(entry)
    daily_limit = _to_nonnegative_int(normalized.get("daily_limit"))
    used = _to_nonnegative_int(normalized.get("used"))
    reserved = _to_nonnegative_int(normalized.get("reserved"))
    return {
        "model_id": str(normalized.get("model_id") or "").strip(),
        "daily_limit": daily_limit,
        "used": used,
        "reserved": reserved,
        "remaining": max(0, daily_limit - used - reserved),
        "date": str(normalized.get("date") or "").strip() or _today_key(),
        "updated_at": normalized.get("updated_at"),
    }


def _third_party_wallet_key(group_id: str, model_id: str) -> str:
    group_id = str(group_id or "").strip()
    model_id = str(model_id or "").strip()
    return f"{group_id}:{model_id}" if group_id and model_id else ""


def _is_matching_third_party_wallet(wallet: dict[str, object], group_id: str, model_id: str) -> bool:
    group_id = str(group_id or "").strip()
    model_id = str(model_id or "").strip()
    wallet_group_id = str(wallet.get("group_id") or "").strip()
    wallet_model_id = str(wallet.get("model_id") or "").strip()
    wallet_id = str(wallet.get("id") or "").strip()
    if wallet_id == _third_party_wallet_key(group_id, model_id):
        return True
    return wallet_group_id == group_id and wallet_model_id == model_id


def _find_third_party_wallet_for_reserve(
    wallets: list[dict[str, object]],
    group_id: str,
    model_id: str,
    amount: int | float,
) -> dict[str, object] | None:
    matches = [wallet for wallet in wallets if _is_matching_third_party_wallet(wallet, group_id, model_id)]
    if not matches:
        return None
    for wallet in matches:
        if _credit_decimal(wallet.get("remaining")) >= _credit_decimal(amount):
            return wallet
    return matches[0]


def _find_third_party_wallet_for_settle(
    wallets: list[dict[str, object]],
    group_id: str,
    model_id: str,
) -> dict[str, object] | None:
    matches = [wallet for wallet in wallets if _is_matching_third_party_wallet(wallet, group_id, model_id)]
    if not matches:
        return None
    for wallet in matches:
        if _credit_decimal(wallet.get("reserved")) > 0:
            return wallet
    return matches[0]


def _third_party_wallet_public(wallet: dict[str, object]) -> dict[str, object]:
    return {
        "id": wallet.get("id"),
        "group_id": wallet.get("group_id"),
        "model_id": wallet.get("model_id"),
        "remaining": _credit_number(wallet.get("remaining")),
        "reserved": _credit_number(wallet.get("reserved")),
        "used": _credit_number(wallet.get("used")),
        "total_recharged": _credit_number(wallet.get("total_recharged")),
        "updated_at": wallet.get("updated_at"),
    }


def _reset_package_periods(grant: dict[str, object], now: datetime | None = None) -> dict[str, object]:
    date_key, week_key, month_key = _period_keys(now)
    next_grant = dict(grant)
    if str(next_grant.get("daily_key") or "") != date_key:
        next_grant["daily_key"] = date_key
        next_grant["used_daily"] = 0
        next_grant["reserved_daily"] = 0
    if str(next_grant.get("weekly_key") or "") != week_key:
        next_grant["weekly_key"] = week_key
        next_grant["used_weekly"] = 0
        next_grant["reserved_weekly"] = 0
    if str(next_grant.get("monthly_key") or "") != month_key:
        next_grant["monthly_key"] = month_key
        next_grant["used_monthly"] = 0
        next_grant["reserved_monthly"] = 0
    return next_grant


def _package_active(grant: dict[str, object]) -> bool:
    starts_at = str(grant.get("starts_at") or "").strip()
    parsed_starts_at = _parse_iso_datetime(starts_at) if starts_at else None
    return (parsed_starts_at is None or parsed_starts_at <= datetime.now(timezone.utc)) and _is_future_datetime(grant.get("expires_at"))


def _package_available(grant: dict[str, object]) -> int | float:
    if not _package_active(grant):
        return 0
    normalized = _reset_package_periods(grant)
    candidates: list[Decimal] = []
    limit_pairs = [
        ("total_limit", "used_total", "reserved_total"),
        ("daily_limit", "used_daily", "reserved_daily"),
        ("weekly_limit", "used_weekly", "reserved_weekly"),
        ("monthly_limit", "used_monthly", "reserved_monthly"),
    ]
    for limit_key, used_key, reserved_key in limit_pairs:
        limit = _to_nonnegative_int(normalized.get(limit_key))
        if limit <= 0:
            continue
        candidates.append(max(Decimal("0"), Decimal(limit) - _credit_decimal(normalized.get(used_key)) - _credit_decimal(normalized.get(reserved_key))))
    if not candidates:
        return PACKAGE_UNLIMITED_SENTINEL
    return _credit_number(min(candidates))


def _active_package_grants(item: dict[str, object]) -> list[dict[str, object]]:
    grants: list[dict[str, object]] = []
    for grant in _normalize_package_grants(item.get("package_grants")):
        grant = _reset_package_periods(grant)
        if _package_active(grant):
            grants.append(grant)
    return grants


def _settleable_package_grants(item: dict[str, object]) -> list[dict[str, object]]:
    return [_reset_package_periods(grant) for grant in _normalize_package_grants(item.get("package_grants"))]


def _package_credit_available(item: dict[str, object]) -> int | float:
    total = Decimal("0")
    for grant in _active_package_grants(item):
        total += _credit_decimal(_package_available(grant))
    return _credit_number(min(total, Decimal(PACKAGE_UNLIMITED_SENTINEL)))


class AuthService:
    # last_used_at 的写盘节流窗口：同一密钥 60 秒内只回写一次（见 authenticate）。
    _TOUCH_INTERVAL_SECONDS = 60.0
    # 后台把「待写入的 last_used_at」批量落库的间隔。
    _TOUCH_FLUSH_INTERVAL_SECONDS = 30.0
    # 缓存未命中时回读整表的节流：未知 token 不能把事件循环拖进数据库。
    _RELOAD_MIN_INTERVAL_SECONDS = 2.0

    def __init__(self, storage: StorageBackend):
        self.storage = storage
        self._lock = RLock()
        self._items = self._load()
        # key_id -> 上次写 last_used_at 的时间戳；只在本进程内存里，重启即清空（无害）。
        self._last_touch_ts: dict[str, float] = {}
        # key_id -> 待落库的 last_used_at（ISO 字符串）。
        # ⚠️⚠️ 为什么不在请求里直接写库：authenticate 大量跑在**事件循环线程**上
        # （async 路由里同步调 require_identity）。以前每次 touch 都要「读整表 →
        # 改一条 → 写整表」，图片任务线程又在并发结算积分抢同一把存储锁，
        # 事件循环排到锁后面 ⇒ 15 秒无响应、整站打不开（2026-10-19 事故）。
        # 现在只改内存 + 记脏，由后台线程批量合并写一次。
        self._pending_touch: dict[str, str] = {}
        self._last_reload_ts = 0.0
        self._reload_requested = False
        self._flusher_started = False

    def _ensure_touch_flusher(self) -> None:
        """惰性启动后台 flush 线程（不在 import 期起，避免测试里凭空多线程）。"""
        if self._flusher_started:
            return
        self._flusher_started = True

        def loop() -> None:
            while True:
                time.sleep(self._TOUCH_FLUSH_INTERVAL_SECONDS)
                try:
                    self._flush_reload_if_requested()
                    self.flush_pending_touches()
                except Exception as exc:  # pragma: no cover - 只记录，不让线程死掉
                    print(f"[auth-service] background flush failed: {exc}")

        thread = Thread(target=loop, name="auth-touch-flusher", daemon=True)
        thread.start()

    def _flush_reload_if_requested(self) -> None:
        """后台兜底：有未命中请求时才回读一次存储（绝不在事件循环里做）。"""
        with self._lock:
            if not self._reload_requested:
                return
            now_ts = time.time()
            if now_ts - self._last_reload_ts < self._RELOAD_MIN_INTERVAL_SECONDS:
                return
            self._last_reload_ts = now_ts
            self._reload_requested = False
            fallback = self._items
        # 读库放在锁外，避免长时间持锁。
        items = self._load(fallback=fallback)
        with self._lock:
            self._items = items

    def flush_pending_touches(self) -> int:
        """把内存里攒下的 last_used_at 一次性写回存储，返回写入条数。"""
        with self._lock:
            pending = dict(self._pending_touch)
            self._pending_touch.clear()
        if not pending:
            return 0

        def mutate(items: list[dict[str, object]]) -> tuple[list[dict[str, object]], int]:
            written = 0
            for index, item in enumerate(items):
                key_id = self._clean(item.get("id"))
                ts = pending.get(key_id)
                if not ts or item.get("last_used_at") == ts:
                    continue
                next_item = dict(item)
                next_item["last_used_at"] = ts
                items[index] = next_item
                written += 1
            return items, written

        try:
            return int(self._update_items_locked(mutate))
        except Exception as exc:
            # 写失败就把脏数据放回去，下一轮再试（不要让统计字段拖垮请求路径）。
            with self._lock:
                for key_id, ts in pending.items():
                    self._pending_touch.setdefault(key_id, ts)
            print(f"[auth-service] flush last_used_at failed: {exc}")
            return 0

    @staticmethod
    def _clean(value: object) -> str:
        return str(value or "").strip()

    @staticmethod
    def _default_name(role: object) -> str:
        normalized = str(role or "").strip().lower()
        if normalized == "admin":
            return "管理员密钥"
        if normalized == "normal":
            return "普通用户"
        return "内部用户"

    def _normalize_item(self, raw: object) -> dict[str, object] | None:
        if not isinstance(raw, dict):
            return None
        role = self._clean(raw.get("role")).lower()
        if role not in {"admin", "user", "normal"}:
            return None
        key_hash = self._clean(raw.get("key_hash"))
        if not key_hash:
            return None
        item_id = self._clean(raw.get("id")) or uuid.uuid4().hex[:12]
        name = self._clean(raw.get("name"))
        if not name and not self._clean(raw.get("wechat_openid")):
            # 微信注册用户允许昵称为空（由用户自己在弹窗里填写），不写任何内置默认名。
            name = self._default_name(role)
        created_at = self._clean(raw.get("created_at")) or _now_iso()
        last_used_at = self._clean(raw.get("last_used_at")) or None
        return {
            "id": item_id,
            "name": name,
            "role": role,
            "key_hash": key_hash,
            "enabled": bool(raw.get("enabled", True)),
            "account_pool_enabled": bool(raw.get("account_pool_enabled", False)),
            "image_memory_button_enabled": bool(raw.get("image_memory_button_enabled", False)),
            "third_party_image_enabled": bool(raw.get("third_party_image_enabled", False)),
            "third_party_image_group_ids": _normalize_string_list(raw.get("third_party_image_group_ids")),
            "third_party_image_model_ids": _normalize_string_list(raw.get("third_party_image_model_ids")),
            "third_party_image_wallets": _normalize_third_party_image_wallets(raw.get("third_party_image_wallets")),
            "image_model_daily_limits": _normalize_image_model_daily_limits(raw.get("image_model_daily_limits")),
            "image_model_daily_default_usages": _normalize_image_model_daily_limits(raw.get("image_model_daily_default_usages")),
            "created_at": created_at,
            "last_used_at": last_used_at,
            "credits": _credit_number(raw.get("credits")),
            # 签到专用账本：签到发放、且尚未被消耗掉的积分余额（见 checkin_admin_summary 注释）。
            # 缺字段一律视为 0 —— 该功能上线前已发放的签到积分无法追溯，故不回溯估算。
            "checkin_credit_remaining": _credit_number(raw.get("checkin_credit_remaining")),
            "reserved_credits": _credit_number(raw.get("reserved_credits")),
            "used_credits": _credit_number(raw.get("used_credits")),
            "total_recharged": _credit_number(raw.get("total_recharged")),
            "credit_grants": _normalize_credit_grants(raw.get("credit_grants")),
            "package_grants": _normalize_package_grants(raw.get("package_grants")),
            "daily_credits_date": self._clean(raw.get("daily_credits_date")),
            "daily_credits_used": _credit_number(raw.get("daily_credits_used")),
            "daily_bonus_credits": _credit_number(raw.get("daily_bonus_credits")),
            "daily_bonus_model_ids": _normalize_string_list(raw.get("daily_bonus_model_ids")),
            "daily_reserved_credits": _credit_number(raw.get("daily_reserved_credits")),
            "api_key_display": self._clean(raw.get("api_key_display")),
            "wechat_openid": self._clean(raw.get("wechat_openid")),
            "wechat_nickname": self._clean(raw.get("wechat_nickname")),
            "wechat_avatar": self._clean(raw.get("wechat_avatar")),
            "avatar_url": self._clean(raw.get("avatar_url")),
            "avatar_update_date": self._clean(raw.get("avatar_update_date")),
            "avatar_update_count": _to_nonnegative_int(raw.get("avatar_update_count")),
            "wechat_bound_at": self._clean(raw.get("wechat_bound_at")) or None,
            "invite_code": self._clean(raw.get("invite_code")),
            "invited_by": self._clean(raw.get("invited_by")),
            "invite_records": [dict(entry) for entry in (raw.get("invite_records") if isinstance(raw.get("invite_records"), list) else []) if isinstance(entry, dict)],
            "total_checkin_rewards": _to_nonnegative_int(raw.get("total_checkin_rewards")),
            "checkin_history": [
                {
                    "date": self._clean(entry.get("date")),
                    "reward": _to_nonnegative_int(entry.get("reward")),
                    "mode": "random" if self._clean(entry.get("mode")) == "random" else "checkin",
                    "created_at": self._clean(entry.get("created_at")),
                }
                for entry in (raw.get("checkin_history") if isinstance(raw.get("checkin_history"), list) else [])
                if isinstance(entry, dict) and self._clean(entry.get("date"))
            ][-400:],
        }

    def _load(self, *, fallback: list[dict[str, object]] | None = None) -> list[dict[str, object]]:
        try:
            items = self.storage.load_auth_keys()
        except Exception as exc:
            print(f"[auth-service] load auth keys failed: {exc}")
            return [dict(item) for item in (fallback or [])]
        return self._normalize_items(items, fallback=fallback)

    def _normalize_items(
        self,
        items: object,
        *,
        fallback: list[dict[str, object]] | None = None,
    ) -> list[dict[str, object]]:
        if not isinstance(items, list):
            return [dict(item) for item in (fallback or [])]
        return [normalized for item in items if (normalized := self._normalize_item(item)) is not None]

    def _save(
        self,
        changed_items: list[dict[str, object]] | None = None,
        deleted_ids: list[str] | None = None,
    ) -> None:
        merge_auth_keys = getattr(self.storage, "merge_auth_keys", None)
        if callable(merge_auth_keys) and (changed_items is not None or deleted_ids):
            merge_auth_keys(
                [dict(item) for item in (changed_items or [])],
                [item_id for item_id in (self._clean(item) for item in (deleted_ids or [])) if item_id],
            )
            return
        self.storage.save_auth_keys(self._items)

    def _reload_locked(self) -> None:
        self._items = self._load(fallback=self._items)

    def _update_items_locked(
        self,
        mutator: Callable[[list[dict[str, object]]], tuple[list[dict[str, object]], ResultT]],
    ) -> ResultT:
        update_auth_keys_locked = getattr(self.storage, "update_auth_keys_locked", None)
        if callable(update_auth_keys_locked):
            def wrapped(raw_items: list[dict[str, object]]) -> tuple[list[dict[str, object]], ResultT]:
                # ⚠️ 这里已经持有**存储**锁（update_auth_keys_locked 内部），再去拿鉴权锁时
                # 必须保证没有「持鉴权锁等存储锁」的线程 —— 这就是 ABBA 死锁的另一半。
                # 现在 authenticate 已改为「先放鉴权锁、再写存储」，加上 list_keys 不再回读
                # 存储，环被打断；锁顺序统一为 存储 → 鉴权。
                with self._lock:
                    current = self._normalize_items(raw_items, fallback=self._items)
                    self._items = current
                next_items, result = mutator(current)
                with self._lock:
                    normalized_next = self._normalize_items(next_items, fallback=current)
                    self._items = normalized_next
                return normalized_next, result

            return update_auth_keys_locked(wrapped)

        with self._lock:
            self._reload_locked()
            next_items, result = mutator(self._items)
            self._items = self._normalize_items(next_items, fallback=self._items)
            self._save()
        return result

    @staticmethod
    def _daily_state(item: dict[str, object]) -> tuple[int | float, int | float]:
        if item.get("role") != "normal":
            return 0, 0
        reserved = _credit_number(item.get("daily_reserved_credits"))
        if AuthService._clean(item.get("daily_credits_date")) != _today_key() and reserved <= 0:
            return 0, 0
        return _credit_number(item.get("daily_credits_used")), reserved

    @staticmethod
    def _daily_bonus_credits(item: dict[str, object]) -> int | float:
        if item.get("role") != "normal":
            return 0
        bonus = _credit_number(item.get("daily_bonus_credits"))
        if bonus <= 0:
            return 0
        if AuthService._clean(item.get("daily_credits_date")) == _today_key():
            return bonus
        if _credit_decimal(item.get("daily_reserved_credits")) > 0:
            return bonus
        return 0

    @staticmethod
    def _daily_bonus_model_ids(item: dict[str, object]) -> list[str]:
        if AuthService._daily_bonus_credits(item) <= 0:
            return []
        model_ids = _normalize_string_list(item.get("daily_bonus_model_ids"))
        # Daily gift codes created before model scoping existed are deliberately
        # kept on the safest default: the built-in account pool only.
        return model_ids or [DAILY_GIFT_ACCOUNT_POOL_MODEL_ID]

    @staticmethod
    def _daily_total(item: dict[str, object]) -> int | float:
        if item.get("role") != "normal":
            return 0
        return _credit_number(Decimal(config.normal_user_daily_credits) + _credit_decimal(AuthService._daily_bonus_credits(item)))

    @staticmethod
    def _daily_remaining(item: dict[str, object], model_id: object = "") -> int | float:
        used, reserved = AuthService._daily_state(item)
        normalized_model_id = AuthService._clean(model_id)
        if not normalized_model_id:
            return _credit_number(max(Decimal("0"), _credit_decimal(AuthService._daily_total(item)) - _credit_decimal(used) - _credit_decimal(reserved)))
        base_remaining = _credit_number(max(Decimal("0"), Decimal(config.normal_user_daily_credits) - _credit_decimal(used) - _credit_decimal(reserved)))
        if normalized_model_id not in AuthService._daily_bonus_model_ids(item):
            return base_remaining
        return _credit_number(max(Decimal("0"), _credit_decimal(AuthService._daily_total(item)) - _credit_decimal(used) - _credit_decimal(reserved)))

    @staticmethod
    def _reset_daily_if_needed(item: dict[str, object]) -> dict[str, object]:
        if item.get("role") != "normal":
            return item
        today = _today_key()
        if AuthService._clean(item.get("daily_credits_date")) == today:
            return item
        if _credit_decimal(item.get("daily_reserved_credits")) > 0:
            return item
        next_item = dict(item)
        next_item["daily_credits_date"] = today
        next_item["daily_credits_used"] = 0
        next_item["daily_bonus_credits"] = 0
        next_item["daily_bonus_model_ids"] = []
        next_item["daily_reserved_credits"] = 0
        return next_item

    @staticmethod
    def _public_item(item: dict[str, object]) -> dict[str, object]:
        permanent_credits = _credit_number(item.get("credits"))
        expiring_credits = _expiring_credit_remaining(item)
        paid_credits = _credit_number(_credit_decimal(permanent_credits) + _credit_decimal(expiring_credits))
        package_credits = _package_credit_available(item)
        daily_used, daily_reserved = AuthService._daily_state(item)
        daily_total = AuthService._daily_total(item)
        daily_bonus = AuthService._daily_bonus_credits(item)
        daily_remaining = AuthService._daily_remaining(item)
        total_available = _credit_number(_credit_decimal(paid_credits) + _credit_decimal(package_credits) + _credit_decimal(daily_remaining)) if item.get("role") == "normal" else paid_credits
        checkin_credits, checkin_credits_expires_at = _checkin_credit_state(item)
        return {
            "id": item.get("id"),
            "name": item.get("name"),
            "role": item.get("role"),
            "enabled": bool(item.get("enabled", True)),
            "account_pool_enabled": bool(item.get("account_pool_enabled", False)),
            "image_memory_button_enabled": bool(item.get("image_memory_button_enabled", False)),
            "third_party_image_enabled": bool(item.get("third_party_image_enabled", False)),
            "third_party_image_group_ids": _normalize_string_list(item.get("third_party_image_group_ids")),
            "third_party_image_model_ids": _normalize_string_list(item.get("third_party_image_model_ids")),
            "third_party_image_wallets": [
                _third_party_wallet_public(wallet)
                for wallet in _normalize_third_party_image_wallets(item.get("third_party_image_wallets"))
                if _to_nonnegative_int(wallet.get("remaining")) + _to_nonnegative_int(wallet.get("reserved")) + _to_nonnegative_int(wallet.get("used")) > 0
            ],
            "image_model_daily_limits": [
                _image_model_daily_limit_public(limit)
                for limit in _normalize_image_model_daily_limits(item.get("image_model_daily_limits"))
            ] if item.get("role") == "user" else [],
            "image_model_daily_default_usages": [
                _image_model_daily_limit_public(limit)
                for limit in _normalize_image_model_daily_limits(item.get("image_model_daily_default_usages"))
            ] if item.get("role") == "user" else [],
            "created_at": item.get("created_at"),
            "last_used_at": item.get("last_used_at"),
            "credits": total_available,
            "paid_credits": paid_credits,
            "permanent_credits": permanent_credits,
            "expiring_credits": expiring_credits,
            "expiring_reserved_credits": _expiring_credit_reserved(item),
            "package_credits": package_credits if item.get("role") == "normal" else 0,
            "package_grants": [
                {
                    "id": grant.get("id"),
                    "name": grant.get("name"),
                    "remaining": _package_available(grant),
                    "created_at": grant.get("created_at"),
                    "starts_at": grant.get("starts_at"),
                    "expires_at": grant.get("expires_at"),
                    "source": grant.get("source"),
                    "code_id": grant.get("code_id"),
                    "daily_limit": _to_nonnegative_int(grant.get("daily_limit")),
                    "weekly_limit": _to_nonnegative_int(grant.get("weekly_limit")),
                    "monthly_limit": _to_nonnegative_int(grant.get("monthly_limit")),
                    "total_limit": _to_nonnegative_int(grant.get("total_limit")),
                    "used_total": _credit_number(grant.get("used_total")),
                    "reserved_total": _credit_number(grant.get("reserved_total")),
                    "used_daily": _credit_number(grant.get("used_daily")),
                    "reserved_daily": _credit_number(grant.get("reserved_daily")),
                    "used_weekly": _credit_number(grant.get("used_weekly")),
                    "reserved_weekly": _credit_number(grant.get("reserved_weekly")),
                    "used_monthly": _credit_number(grant.get("used_monthly")),
                    "reserved_monthly": _credit_number(grant.get("reserved_monthly")),
                }
                for grant in _active_package_grants(item)
                if _package_available(grant) > 0
            ],
            "credit_grants": [
                {
                    "id": grant.get("id"),
                    "remaining": _credit_number(grant.get("remaining")),
                    "reserved": _credit_number(grant.get("reserved")),
                    "expires_at": grant.get("expires_at"),
                    "source": grant.get("source"),
                }
                for grant in _active_credit_grants(item)
                if _credit_decimal(grant.get("remaining")) + _credit_decimal(grant.get("reserved")) > 0
            ],
            "reserved_credits": _credit_number(item.get("reserved_credits")),
            "used_credits": _credit_number(item.get("used_credits")),
            "total_recharged": _credit_number(item.get("total_recharged")),
            "checkin_credit_remaining": _credit_number(item.get("checkin_credit_remaining")),
            # 「未使用的签到积分」= 永久形态账本 + 未过期的签到券；expires_at 为最近到期时间（空 = 永久/无）。
            "checkin_credits": _credit_number(checkin_credits),
            "checkin_credits_expires_at": checkin_credits_expires_at,
            # 逐笔到期明细（按到期升序；永久形态为空列表），供前端逐行展示「x 积分 N 天后到期」。
            "checkin_credit_batches": _checkin_credit_batches(item),
            "daily_credits": daily_total if item.get("role") == "normal" else 0,
            "daily_credits_remaining": daily_remaining,
            "daily_credits_used": daily_used,
            "daily_bonus_credits": daily_bonus,
            "daily_bonus_model_ids": AuthService._daily_bonus_model_ids(item),
            "daily_reserved_credits": daily_reserved,
            "daily_credits_date": _today_key() if item.get("role") == "normal" else "",
            "has_api_key_display": bool(AuthService._clean(item.get("api_key_display"))),
            "wechat_openid": AuthService._clean(item.get("wechat_openid")),
            "wechat_nickname": AuthService._clean(item.get("wechat_nickname")),
            "wechat_avatar": AuthService._clean(item.get("wechat_avatar")),
            "avatar_url": AuthService._clean(item.get("avatar_url")),
            "avatar_update_date": AuthService._clean(item.get("avatar_update_date")),
            "avatar_update_count": _to_nonnegative_int(item.get("avatar_update_count")),
            "wechat_bound_at": item.get("wechat_bound_at"),
            "invite_code": AuthService._clean(item.get("invite_code")),
            "invite_records": [dict(entry) for entry in item.get("invite_records", []) if isinstance(entry, dict)],
            "total_checkin_rewards": _to_nonnegative_int(item.get("total_checkin_rewards")),
        }

    def list_keys(self, role: AuthRole | None = None) -> list[dict[str, object]]:
        # ⚠️⚠️ 这里**绝对不能**再调 _reload_locked()（那会去拿存储锁）。
        # 调用方常常是「已经持有存储事务锁」的商店逻辑（兑换卡密 / 每日礼包在算可用模型），
        # 一旦这里回头去要存储锁，就和 authenticate（持鉴权锁去写存储）形成 ABBA 死锁，
        # 而 authenticate 是跑在**事件循环线程**上的 ⇒ 整站彻底无响应、连接被关闭。
        # 2026-10-19 事故根因。单进程内所有写入都会同步更新 self._items，缓存即权威。
        with self._lock:
            items = [item for item in self._items if role is None or item.get("role") == role]
            return [self._public_item(item) for item in items]

    def checkin_summary(self, user_id: str) -> dict[str, object]:
        normalized_id = self._clean(user_id)
        today = _today_key()
        with self._lock:
            self._reload_locked()
            current = next((item for item in self._items if item.get("id") == normalized_id and item.get("role") == "normal"), None)
            if current is None:
                raise ValueError("普通用户不存在")
            if not self._clean(current.get("invite_code")):
                current["invite_code"] = _invite_code()
                while any(self._clean(item.get("invite_code")) == current["invite_code"] and item.get("id") != current.get("id") for item in self._items):
                    current["invite_code"] = _invite_code()
                self._save(changed_items=[current])
            history = [entry for entry in current.get("checkin_history", []) if isinstance(entry, dict)]
            own = [dict(entry) for entry in history if str(entry.get("date") or "")]
            records = [{"date": str(entry.get("date")), "reward": _to_nonnegative_int(entry.get("reward")), "mode": str(entry.get("mode") or "checkin")} for entry in own]
            users: list[dict[str, object]] = []
            for item in self._items:
                if item.get("role") != "normal":
                    continue
                entries = [entry for entry in item.get("checkin_history", []) if isinstance(entry, dict)]
                users.append({
                    "id": item.get("id"),
                    "name": item.get("name") or "普通用户",
                    "avatar_url": self._clean(item.get("avatar_url")) or self._clean(item.get("wechat_avatar")),
                    "wechat_avatar": self._clean(item.get("wechat_avatar")),
                    "history": entries,
                })
            daily: list[dict[str, object]] = []
            total: list[dict[str, object]] = []
            for user in users:
                entries = user["history"] if isinstance(user.get("history"), list) else []
                today_entries = [entry for entry in entries if isinstance(entry, dict) and str(entry.get("date") or "") == today]
                today_reward = sum(_to_nonnegative_int(entry.get("reward")) for entry in today_entries)
                today_created_at = min((str(entry.get("created_at") or "") for entry in today_entries), default="")
                total_reward = sum(_to_nonnegative_int(entry.get("reward")) for entry in entries if isinstance(entry, dict))
                days = len({str(entry.get("date")) for entry in entries if isinstance(entry, dict) and str(entry.get("date") or "")})
                if today_reward:
                    daily.append({
                        "name": user["name"],
                        "avatar_url": user.get("avatar_url") or "",
                        "wechat_avatar": user.get("wechat_avatar") or "",
                        "reward": today_reward,
                        "days": days,
                        "created_at": today_created_at,
                    })
                if total_reward:
                    total.append({
                        "name": user["name"],
                        "avatar_url": user.get("avatar_url") or "",
                        "wechat_avatar": user.get("wechat_avatar") or "",
                        "reward": total_reward,
                        "days": days,
                    })
            daily.sort(key=lambda item: (str(item.get("created_at") or ""), str(item["name"])))
            total.sort(key=lambda item: (-int(item["days"]), str(item["name"])))
            invite_records = []
            for entry in current.get("invite_records", []):
                if not isinstance(entry, dict):
                    continue
                record = dict(entry)
                invitee_id = self._clean(record.get("invitee_id"))
                invitee = next((user for user in users if self._clean(user.get("id")) == invitee_id), None)
                if invitee is not None:
                    record["invitee_name"] = invitee.get("name") or record.get("invitee_name")
                invite_records.append(record)
            total_rewards = sum(_to_nonnegative_int(entry.get("reward")) for entry in own)
            invite_rewards = sum(_to_nonnegative_int(entry.get("reward")) for entry in invite_records)
            return {
                "today": today,
                "checked_in": any(str(entry.get("date")) == today for entry in own),
                "records": records,
                "daily": daily[:100],
                "total": total[:100],
                "total_rewards": total_rewards,
                "invite": {
                    "code": self._clean(current.get("invite_code")),
                    "successful": len(invite_records),
                    "rewards": invite_rewards,
                    "records": invite_records,
                },
            }

    def checkin_admin_summary(self) -> dict[str, object]:
        today = _today_key()
        with self._lock:
            self._reload_locked()
            users = [item for item in self._items if item.get("role") == "normal"]
            today_count = 0
            total_rewards = 0
            unused_credits = Decimal("0.00")
            for item in users:
                entries = [entry for entry in item.get("checkin_history", []) if isinstance(entry, dict)]
                if any(str(entry.get("date") or "") == today for entry in entries):
                    today_count += 1
                issued = sum(_to_nonnegative_int(entry.get("reward")) for entry in entries)
                total_rewards += issued
                # 「未使用」= 签到专用账本（永久形态）+ 未过期的签到券（限时形态，见 _checkin_credit_state）。
                # 账本只统计本功能上线后的签到发放 —— 上线前已发放的签到积分无法追溯，不回溯估算；
                # 因此上线初期这个数会低于 total_rewards，属预期。
                checkin_unused, _ = _checkin_credit_state(item)
                unused_credits += checkin_unused
            return {
                "normal_users": len(users),
                "today_checked_in": today_count,
                "total_rewards": total_rewards,
                "unused_credits": _credit_number(unused_credits),
            }

    def perform_checkin(self, user_id: str, *, mode: str = "checkin", reward: int = 1) -> dict[str, object]:
        normalized_id = self._clean(user_id)
        today = _today_key()
        normalized_mode = "random" if mode == "random" else "checkin"
        normalized_reward = max(1, min(1000, int(reward or 1)))
        # 后台「签到积分可用天数」：0 = 永久（进基础积分池）；>0 则发放成带 expires_at 的限时券。
        valid_days = max(0, min(3650, int(config.checkin_credit_valid_days or 0)))
        expires_at = (datetime.now(timezone.utc) + timedelta(days=valid_days)).isoformat() if valid_days > 0 else ""
        def mutate(items: list[dict[str, object]]) -> tuple[list[dict[str, object]], dict[str, object]]:
            for index, item in enumerate(items):
                if item.get("id") != normalized_id or item.get("role") != "normal":
                    continue
                history = [dict(entry) for entry in item.get("checkin_history", []) if isinstance(entry, dict)]
                if any(str(entry.get("date") or "") == today for entry in history):
                    raise ValueError("今天已经签到过了")
                entry = {"date": today, "reward": normalized_reward, "mode": normalized_mode, "created_at": _now_iso()}
                if expires_at:
                    entry["expires_at"] = expires_at
                next_item = dict(item)
                next_item["checkin_history"] = [*history[-399:], entry]
                next_items = list(items)
                next_items[index] = next_item
                return next_items, {"date": today, "reward": normalized_reward, "mode": normalized_mode, "expires_at": expires_at}
            raise ValueError("普通用户不存在")
        result = self._update_items_locked(mutate)
        self.adjust_credits(normalized_id, normalized_reward, source="checkin", expires_at=expires_at or None)
        return result

    def reconcile_checkin_credit_expiry(self, valid_days: int) -> dict[str, object]:
        """按后台「签到积分可用天数」重算所有普通用户**已发放**的签到积分形态。
        由后台保存该配置时调用（且仅在值真的变化时），让新设置立即对历史积分生效，
        不必等用户下一次签到。语义：
          - valid_days > 0：把永久账本余额（`checkin_credit_remaining`）从基础积分池扣出，
            转成 `expires_at = now + N 天` 的限时券（这部分原来没有到期时间，给足 N 天宽限）；
            已有的签到限时券**保留各自原有节奏**，到期时间 = `created_at + N 天`（缺 created_at 才退回 now + N 天）。
            若重算后已经过了到期时刻（把 N 调小到早于某些批次的发放日），这批直接作废。
          - valid_days == 0：把所有签到限时券的剩余量转回基础积分池 + 永久账本，券移除。
        只处理 `source="checkin"` 的积分；卡密 / 后台加积分 / 邀请 / 套餐券一律不动。
        `reserved > 0`（正在预扣）的券跳过，避免打断进行中的生成任务。
        """
        days = max(0, min(3650, int(valid_days or 0)))
        now = datetime.now(timezone.utc)
        expires_at = (now + timedelta(days=days)).isoformat() if days > 0 else ""
        converted = Decimal("0")
        skipped = 0
        expired = 0
        touched = 0

        def mutate(items: list[dict[str, object]]) -> tuple[list[dict[str, object]], None]:
            nonlocal converted, skipped, expired, touched
            next_items: list[dict[str, object]] = []
            for item in items:
                if item.get("role") != "normal":
                    next_items.append(item)
                    continue
                grants = _normalize_credit_grants(item.get("credit_grants"))
                checkin_grants = [grant for grant in grants if str(grant.get("source") or "").strip() == CHECKIN_CREDIT_SOURCE]
                other_grants = [grant for grant in grants if str(grant.get("source") or "").strip() != CHECKIN_CREDIT_SOURCE]
                credits = _credit_decimal(item.get("credits"))
                ledger = _credit_decimal(item.get("checkin_credit_remaining"))
                next_credits = credits
                next_ledger = ledger
                next_checkin_grants: list[dict[str, object]] = []

                for grant in checkin_grants:
                    if _credit_decimal(grant.get("reserved")) > 0:
                        skipped += 1
                        next_checkin_grants.append(grant)
                        continue
                    remaining = _credit_decimal(grant.get("remaining"))
                    if remaining <= 0:
                        continue
                    if days > 0:
                        # 各批次保留自己的发放节奏：到期 = 发放时刻 + N 天。
                        issued = _parse_iso_datetime(grant.get("created_at"))
                        target = (issued + timedelta(days=days)) if issued is not None else (now + timedelta(days=days))
                        if target <= now:
                            # 按「自发放起 N 天」的口径，这批已经过期 —— 作废（不是扣费，是自然到期）。
                            expired += 1
                            continue
                        next_checkin_grants.append({**grant, "expires_at": target.isoformat()})
                    else:
                        next_credits += remaining
                        next_ledger += remaining
                        converted += remaining

                if days > 0:
                    movable = min(ledger, credits)
                    if movable > 0:
                        next_credits = credits - movable
                        next_ledger = ledger - movable
                        next_checkin_grants.append(
                            {
                                "id": uuid.uuid4().hex[:12],
                                "remaining": _credit_number(movable),
                                "reserved": 0,
                                "expires_at": expires_at,
                                "created_at": _now_iso(),
                                "source": CHECKIN_CREDIT_SOURCE,
                                "code_id": "",
                            }
                        )
                        converted += movable

                next_grants = [*other_grants, *next_checkin_grants]
                next_grants.sort(key=lambda grant: str(grant.get("expires_at") or ""))
                if next_credits == credits and next_ledger == ledger and next_grants == grants:
                    next_items.append(item)
                    continue
                next_item = dict(item)
                next_item["credits"] = _credit_number(next_credits)
                next_item["checkin_credit_remaining"] = _credit_number(next_ledger)
                next_item["credit_grants"] = next_grants
                next_items.append(next_item)
                touched += 1
            return next_items, None

        self._update_items_locked(mutate)
        result = {
            "valid_days": days,
            "users": touched,
            "converted": _credit_number(converted),
            "expired_grants": expired,
            "skipped_grants": skipped,
        }
        print(f"[checkin] 签到积分有效期重算：{result}")
        return result

    def get_key(self, key_id: str, *, role: AuthRole | None = None) -> dict[str, object] | None:
        normalized_id = self._clean(key_id)
        if not normalized_id:
            return None
        with self._lock:
            self._reload_locked()
            for item in self._items:
                if item.get("id") == normalized_id and (role is None or item.get("role") == role):
                    return self._public_item(item)
        return None

    def _has_key_hash_locked(self, key_hash: str, *, exclude_id: str = "") -> bool:
        for item in self._items:
            item_id = self._clean(item.get("id"))
            if exclude_id and item_id == exclude_id:
                continue
            stored_hash = self._clean(item.get("key_hash"))
            if stored_hash and hmac.compare_digest(stored_hash, key_hash):
                return True
        return False

    def _build_key_hash_locked(self, raw_key: str, *, exclude_id: str = "") -> str:
        candidate = self._clean(raw_key)
        if not candidate:
            raise ValueError("请输入新的专用密钥")
        admin_key = self._clean(config.auth_key)
        if admin_key and hmac.compare_digest(candidate, admin_key):
            raise ValueError("这个密钥和管理员密钥冲突，请换一个")
        key_hash = _hash_key(candidate)
        if self._has_key_hash_locked(key_hash, exclude_id=exclude_id):
            raise ValueError("这个专用密钥已经存在，请换一个")
        return key_hash

    def _has_name_locked(self, name: str, *, role: AuthRole | None = None, exclude_id: str = "") -> bool:
        candidate = self._clean(name)
        if not candidate:
            return False
        for item in self._items:
            item_id = self._clean(item.get("id"))
            if exclude_id and item_id == exclude_id:
                continue
            if role is not None and item.get("role") != role:
                continue
            if self._clean(item.get("name")) == candidate:
                return True
        return False

    def _build_default_name_locked(self, role: AuthRole, *, exclude_id: str = "") -> str:
        base_name = self._default_name(role)
        if not self._has_name_locked(base_name, role=role, exclude_id=exclude_id):
            return base_name
        suffix = 2
        while True:
            candidate = f"{base_name} {suffix}"
            if not self._has_name_locked(candidate, role=role, exclude_id=exclude_id):
                return candidate
            suffix += 1

    def _build_name_locked(self, name: str, *, role: AuthRole, exclude_id: str = "") -> str:
        candidate = self._clean(name)
        if not candidate:
            return self._build_default_name_locked(role, exclude_id=exclude_id)
        if len(candidate) > 32:
            raise ValueError("昵称最多 32 个字符")
        if self._has_name_locked(candidate, role=role, exclude_id=exclude_id):
            raise ValueError("这个名称已经在使用中了，请换一个")
        return candidate

    def _build_unique_name_locked(self, name: str, *, role: AuthRole, exclude_id: str = "") -> str:
        candidate = self._clean(name)
        if not candidate:
            return self._build_default_name_locked(role, exclude_id=exclude_id)
        if not self._has_name_locked(candidate, role=role, exclude_id=exclude_id):
            return candidate
        suffix = 2
        while True:
            next_candidate = f"{candidate} {suffix}"
            if not self._has_name_locked(next_candidate, role=role, exclude_id=exclude_id):
                return next_candidate
            suffix += 1

    def create_key(self, *, role: AuthRole, name: str = "") -> tuple[dict[str, object], str]:
        def mutate(items: list[dict[str, object]]) -> tuple[list[dict[str, object]], tuple[dict[str, object], str]]:
            normalized_name = self._build_name_locked(name, role=role)
            while True:
                raw_key = f"sk-{secrets.token_urlsafe(24)}"
                try:
                    key_hash = self._build_key_hash_locked(raw_key)
                    break
                except ValueError:
                    continue
            item = {
                "id": uuid.uuid4().hex[:12],
                "name": normalized_name,
                "role": role,
                "key_hash": key_hash,
                "enabled": True,
                "account_pool_enabled": False,
                "image_memory_button_enabled": False,
                "third_party_image_enabled": False,
                "third_party_image_group_ids": [],
                "third_party_image_model_ids": [],
                "third_party_image_wallets": [],
                "image_model_daily_limits": [],
                "image_model_daily_default_usages": [],
                "created_at": _now_iso(),
                "last_used_at": None,
                "credits": config.normal_user_default_credits if role == "normal" else 0,
                "reserved_credits": 0,
                "used_credits": 0,
                "total_recharged": 0,
                "credit_grants": [],
                "package_grants": [],
                "daily_credits_date": _today_key() if role == "normal" else "",
                "daily_credits_used": 0,
                "daily_bonus_credits": 0,
                "daily_reserved_credits": 0,
                "api_key_display": raw_key if role in {"normal", "user"} else "",
                "wechat_openid": "",
                "wechat_nickname": "",
                "wechat_avatar": "",
                "wechat_bound_at": None,
                "checkin_history": [],
                "invite_code": "",
                "invited_by": "",
                "invite_records": [],
                "total_checkin_rewards": 0,
            }
            if role == "normal":
                candidate = _invite_code()
                while any(self._clean(existing.get("invite_code")) == candidate for existing in items):
                    candidate = _invite_code()
                item["invite_code"] = candidate
            return [*items, item], (self._public_item(item), raw_key)

        return self._update_items_locked(mutate)

    def create_invited_normal_user(self, *, name: str, invite_code: str) -> tuple[dict[str, object], str]:
        normalized_name = self._clean(name) or "普通用户"
        normalized_invite_code = self._clean(invite_code).upper()
        if not normalized_invite_code:
            raise ValueError("请输入邀请码")
        invite_reward = config.checkin_invite_reward

        def mutate(items: list[dict[str, object]]) -> tuple[list[dict[str, object]], tuple[dict[str, object], str]]:
            inviter = next((item for item in items if item.get("role") == "normal" and self._clean(item.get("invite_code")).upper() == normalized_invite_code), None)
            if inviter is None:
                raise ValueError("邀请码无效")
            while True:
                raw_key = f"sk-{secrets.token_urlsafe(24)}"
                key_hash = self._build_key_hash_locked(raw_key)
                break
            candidate = _invite_code()
            while any(self._clean(existing.get("invite_code")) == candidate for existing in items):
                candidate = _invite_code()
            item = {
                "id": uuid.uuid4().hex[:12], "name": self._build_unique_name_locked(normalized_name, role="normal"), "role": "normal",
                "key_hash": key_hash, "enabled": True, "account_pool_enabled": False,
                "image_memory_button_enabled": False,
                "third_party_image_enabled": False, "third_party_image_group_ids": [], "third_party_image_model_ids": [],
                "third_party_image_wallets": [], "image_model_daily_limits": [], "image_model_daily_default_usages": [],
                "created_at": _now_iso(), "last_used_at": None, "credits": config.normal_user_default_credits,
                "reserved_credits": 0, "used_credits": 0, "total_recharged": 0, "credit_grants": [], "package_grants": [],
                "daily_credits_date": _today_key(), "daily_credits_used": 0, "daily_bonus_credits": 0, "daily_reserved_credits": 0,
                "api_key_display": raw_key, "wechat_openid": "", "wechat_nickname": "", "wechat_avatar": "", "wechat_bound_at": None,
                "checkin_history": [], "invite_code": candidate, "invited_by": str(inviter.get("id") or ""), "invite_records": [], "total_checkin_rewards": 0,
            }
            record = {"invitee_id": item["id"], "invitee_name": item["name"], "registered_at": _now_iso(), "reward": invite_reward}
            inviter_next = dict(inviter)
            inviter_next["invite_records"] = [*([dict(entry) for entry in inviter.get("invite_records", []) if isinstance(entry, dict)])[-99:], record]
            next_items = list(items)
            next_items[next((i for i, current in enumerate(items) if current.get("id") == inviter.get("id")), 0)] = inviter_next
            next_items.append(item)
            return next_items, (self._public_item(item), raw_key)

        result = self._update_items_locked(mutate)
        inviter_id = ""
        created = result[0]
        with self._lock:
            self._reload_locked()
            inviter_id = str(next((item.get("id") for item in self._items if self._clean(item.get("invite_code")).upper() == normalized_invite_code), "") or "")
        if inviter_id:
            self.adjust_credits(inviter_id, invite_reward, source="invite")
        self.adjust_credits(str(created.get("id") or ""), invite_reward, source="invite")
        return result

    def attach_invite(self, user_id: str, invite_code: str) -> None:
        normalized_id = self._clean(user_id)
        normalized_code = self._clean(invite_code).upper()
        if not normalized_id or not normalized_code:
            return
        inviter_id = ""
        already = False
        with self._lock:
            self._reload_locked()
            target = next((item for item in self._items if item.get("id") == normalized_id and item.get("role") == "normal"), None)
            inviter = next((item for item in self._items if item.get("role") == "normal" and self._clean(item.get("invite_code")).upper() == normalized_code), None)
            if target is None or inviter is None or target.get("id") == inviter.get("id") or self._clean(target.get("invited_by")):
                return
            inviter_id = str(inviter.get("id") or "")
            invite_reward = config.checkin_invite_reward
            record = {"invitee_id": target.get("id"), "invitee_name": target.get("name"), "registered_at": _now_iso(), "reward": invite_reward}
            next_items = list(self._items)
            target_next = dict(target); target_next["invited_by"] = inviter_id
            inviter_next = dict(inviter); inviter_next["invite_records"] = [*([dict(entry) for entry in inviter.get("invite_records", []) if isinstance(entry, dict)])[-99:], record]
            next_items[next(i for i, item in enumerate(self._items) if item.get("id") == target.get("id"))] = target_next
            next_items[next(i for i, item in enumerate(self._items) if item.get("id") == inviter.get("id"))] = inviter_next
            self._items = self._normalize_items(next_items, fallback=self._items)
            self._save(changed_items=[target_next, inviter_next])
        self.adjust_credits(inviter_id, config.checkin_invite_reward, source="invite")
        self.adjust_credits(normalized_id, config.checkin_invite_reward, source="invite")

    def get_or_create_wechat_normal_user(
        self,
        *,
        openid: str,
        nickname: str = "",
        avatar: str = "",
        allow_create: bool = True,
    ) -> tuple[dict[str, object], str, bool]:
        normalized_openid = self._clean(openid)
        if not normalized_openid:
            raise ValueError("wechat openid is required")
        normalized_nickname = self._clean(nickname)
        normalized_avatar = self._clean(avatar)
        now = _now_iso()

        def mutate(items: list[dict[str, object]]) -> tuple[list[dict[str, object]], tuple[dict[str, object], str, bool]]:
            for index, item in enumerate(items):
                if item.get("role") != "normal" or self._clean(item.get("wechat_openid")) != normalized_openid:
                    continue
                next_item = dict(item)
                next_item["wechat_nickname"] = normalized_nickname
                next_item["wechat_avatar"] = normalized_avatar
                next_item["wechat_bound_at"] = self._clean(next_item.get("wechat_bound_at")) or now
                raw_key = self._clean(next_item.get("api_key_display"))
                if not raw_key:
                    while True:
                        raw_key = f"sk-{secrets.token_urlsafe(24)}"
                        try:
                            next_item["key_hash"] = self._build_key_hash_locked(raw_key, exclude_id=self._clean(next_item.get("id")))
                            break
                        except ValueError:
                            continue
                    next_item["api_key_display"] = raw_key
                next_items = list(items)
                next_items[index] = next_item
                return next_items, (self._public_item(next_item), raw_key, False)

            if not allow_create:
                raise ValueError("当前暂不开放新用户注册，请使用已绑定过的账号登录。")

            # 新注册的微信用户不预置「微信用户 xxxx」这类内置昵称，留空由用户自己设置。
            normalized_name = ""
            while True:
                raw_key = f"sk-{secrets.token_urlsafe(24)}"
                try:
                    key_hash = self._build_key_hash_locked(raw_key)
                    break
                except ValueError:
                    continue
            item = {
                "id": uuid.uuid4().hex[:12],
                "name": normalized_name,
                "role": "normal",
                "key_hash": key_hash,
                "enabled": True,
                "account_pool_enabled": False,
                "image_memory_button_enabled": False,
                "third_party_image_enabled": False,
                "third_party_image_group_ids": [],
                "third_party_image_model_ids": [],
                "third_party_image_wallets": [],
                "created_at": now,
                "last_used_at": None,
                "credits": config.normal_user_default_credits,
                "reserved_credits": 0,
                "used_credits": 0,
                "total_recharged": 0,
                "credit_grants": [],
                "package_grants": [],
                "daily_credits_date": _today_key(),
                "daily_credits_used": 0,
                "daily_bonus_credits": 0,
                "daily_reserved_credits": 0,
                "api_key_display": raw_key,
                "wechat_openid": normalized_openid,
                "wechat_nickname": normalized_nickname,
                "wechat_avatar": normalized_avatar,
                "wechat_bound_at": now,
                "checkin_history": [],
                "invite_code": _invite_code(),
                "invited_by": "",
                "invite_records": [],
                "total_checkin_rewards": 0,
            }
            return [*items, item], (self._public_item(item), raw_key, True)

        return self._update_items_locked(mutate)

    def get_profile(self, key_id: str) -> dict[str, object] | None:
        normalized_id = self._clean(key_id)
        if not normalized_id:
            return None
        item = self.get_key(normalized_id)
        if item is None:
            return None
        return {
            "id": item.get("id"),
            "name": item.get("name"),
            "role": item.get("role"),
            "created_at": item.get("created_at"),
            "account_pool_enabled": bool(item.get("account_pool_enabled")),
            "third_party_image_enabled": bool(item.get("third_party_image_enabled")),
            "third_party_image_group_ids": _normalize_string_list(item.get("third_party_image_group_ids")),
            "third_party_image_model_ids": _normalize_string_list(item.get("third_party_image_model_ids")),
            "third_party_image_wallets": item.get("third_party_image_wallets") if isinstance(item.get("third_party_image_wallets"), list) else [],
            "wechat_openid": self._clean(item.get("wechat_openid")),
            "wechat_avatar": self._clean(item.get("wechat_avatar")),
            "avatar_url": self._clean(item.get("avatar_url")),
            "avatar_updates_remaining": max(
                0,
                3 - (
                    _to_nonnegative_int(item.get("avatar_update_count"))
                    if self._clean(item.get("avatar_update_date")) == _today_key()
                    else 0
                ),
            ),
            "wechat_bound_at": item.get("wechat_bound_at"),
            "invite_code": self._clean(item.get("invite_code")),
            "invite_records": [dict(entry) for entry in item.get("invite_records", []) if isinstance(entry, dict)],
            "total_checkin_rewards": _to_nonnegative_int(item.get("total_checkin_rewards")),
        }

    def update_self_name(self, key_id: str, name: str) -> dict[str, object]:
        item = self.update_key(key_id, {"name": name}, role="normal")
        if item is None:
            raise ValueError("普通用户不存在")
        return item

    def update_self_avatar(self, key_id: str, avatar_url: str) -> dict[str, object]:
        normalized_id = self._clean(key_id)
        normalized_url = self._clean(avatar_url)
        if not normalized_id or not normalized_url:
            raise ValueError("头像地址无效")
        today = _today_key()

        def mutate(items: list[dict[str, object]]) -> tuple[list[dict[str, object]], dict[str, object]]:
            for index, item in enumerate(items):
                if item.get("id") != normalized_id or item.get("role") != "normal":
                    continue
                update_count = (
                    _to_nonnegative_int(item.get("avatar_update_count"))
                    if self._clean(item.get("avatar_update_date")) == today
                    else 0
                )
                if update_count >= 3:
                    raise ValueError("头像每天最多修改 3 次，请明天再试")
                next_item = dict(item)
                next_item["avatar_url"] = normalized_url
                next_item["avatar_update_date"] = today
                next_item["avatar_update_count"] = update_count + 1
                next_items = list(items)
                next_items[index] = next_item
                return next_items, self._public_item(next_item)
            raise ValueError("普通用户不存在")

        return self._update_items_locked(mutate)

    def get_api_key_display(self, key_id: str) -> str:
        normalized_id = self._clean(key_id)
        if not normalized_id:
            return ""
        with self._lock:
            self._reload_locked()
            for item in self._items:
                if item.get("id") == normalized_id and item.get("role") in {"normal", "user"}:
                    return self._clean(item.get("api_key_display"))
        return ""

    def update_key(
        self,
        key_id: str,
        updates: dict[str, object],
        *,
        role: AuthRole | None = None,
    ) -> dict[str, object] | None:
        normalized_id = self._clean(key_id)
        if not normalized_id:
            return None
        def mutate(items: list[dict[str, object]]) -> tuple[list[dict[str, object]], dict[str, object] | None]:
            for index, item in enumerate(items):
                if item.get("id") != normalized_id:
                    continue
                if role is not None and item.get("role") != role:
                    return items, None
                next_item = dict(item)
                next_role = str(next_item.get("role") or "user")
                if next_role not in {"admin", "user", "normal"}:
                    next_role = "user"
                if "name" in updates and updates.get("name") is not None:
                    next_item["name"] = self._build_name_locked(
                        str(updates.get("name") or ""),
                        role=next_role,  # type: ignore[arg-type]
                        exclude_id=normalized_id,
                    )
                if "enabled" in updates and updates.get("enabled") is not None:
                    next_item["enabled"] = bool(updates.get("enabled"))
                if "account_pool_enabled" in updates and updates.get("account_pool_enabled") is not None:
                    next_item["account_pool_enabled"] = bool(updates.get("account_pool_enabled"))
                if "image_memory_button_enabled" in updates and updates.get("image_memory_button_enabled") is not None:
                    next_item["image_memory_button_enabled"] = bool(updates.get("image_memory_button_enabled"))
                if "third_party_image_enabled" in updates and updates.get("third_party_image_enabled") is not None:
                    next_item["third_party_image_enabled"] = bool(updates.get("third_party_image_enabled"))
                if "third_party_image_group_ids" in updates and updates.get("third_party_image_group_ids") is not None:
                    next_item["third_party_image_group_ids"] = _normalize_string_list(updates.get("third_party_image_group_ids"))
                if "third_party_image_model_ids" in updates and updates.get("third_party_image_model_ids") is not None:
                    next_item["third_party_image_model_ids"] = _normalize_string_list(updates.get("third_party_image_model_ids"))
                if "image_model_daily_limits" in updates and updates.get("image_model_daily_limits") is not None:
                    next_item["image_model_daily_limits"] = (
                        _normalize_image_model_daily_limits(updates.get("image_model_daily_limits"))
                        if next_role == "user"
                        else []
                    )
                if "key" in updates and updates.get("key") is not None:
                    raw_key = str(updates.get("key") or "")
                    next_item["key_hash"] = self._build_key_hash_locked(raw_key, exclude_id=normalized_id)
                    if next_role in {"normal", "user"}:
                        next_item["api_key_display"] = self._clean(raw_key)
                next_items = list(items)
                next_items[index] = next_item
                return next_items, self._public_item(next_item)
            return items, None

        return self._update_items_locked(mutate)

    def reserve_internal_image_model_daily_limit(
        self,
        key_id: str,
        *,
        model_id: str,
        amount: int,
    ) -> dict[str, object]:
        normalized_id = self._clean(key_id)
        normalized_model_id = self._clean(model_id)
        normalized_amount = max(0, int(amount or 0))
        if not normalized_id or not normalized_model_id or normalized_amount <= 0:
            return {"_internal_image_model_daily_limit_reservation": {"amount": 0}}

        def mutate(items: list[dict[str, object]]) -> tuple[list[dict[str, object]], dict[str, object]]:
            for index, item in enumerate(items):
                if item.get("id") != normalized_id or item.get("role") != "user":
                    continue
                limits = _normalize_image_model_daily_limits(item.get("image_model_daily_limits"))
                limit = next((entry for entry in limits if self._clean(entry.get("model_id")) == normalized_model_id), None)
                source_field = "image_model_daily_limits"
                reservation_scope = "user"
                limit_scope = "per_user"
                if limit is None:
                    default_limit = config.get_internal_image_model_daily_limit(normalized_model_id)
                    if default_limit is None:
                        return items, {"_internal_image_model_daily_limit_reservation": {"amount": 0}}
                    limit_scope = str(default_limit.get("limit_scope") or "per_user").strip().lower()
                    if limit_scope == "total":
                        total_result = config.reserve_internal_image_model_total_daily_limit(normalized_model_id, normalized_amount)
                        total_reservation = total_result.get("_internal_image_model_daily_limit_reservation") if isinstance(total_result, dict) else {}
                        return items, {
                            "_internal_image_model_daily_limit_reservation": total_reservation
                            if isinstance(total_reservation, dict)
                            else {"amount": 0}
                        }
                    source_field = "image_model_daily_default_usages"
                    reservation_scope = "default_user"
                    limits = _normalize_image_model_daily_limits(item.get(source_field))
                    limit = next((entry for entry in limits if self._clean(entry.get("model_id")) == normalized_model_id), None)
                    if limit is None:
                        limit = {
                            "model_id": normalized_model_id,
                            "daily_limit": _to_nonnegative_int(default_limit.get("daily_limit")),
                            "date": _today_key(),
                            "used": 0,
                            "reserved": 0,
                            "updated_at": "",
                        }
                        limits.append(limit)
                    else:
                        limit["daily_limit"] = _to_nonnegative_int(default_limit.get("daily_limit"))
                limit = _reset_image_model_daily_limit(limit)
                daily_limit = _to_nonnegative_int(limit.get("daily_limit"))
                used = _to_nonnegative_int(limit.get("used"))
                reserved = _to_nonnegative_int(limit.get("reserved"))
                remaining = max(0, daily_limit - used - reserved)
                if normalized_amount > remaining:
                    if remaining <= 0:
                        raise ValueError(f"{normalized_model_id} 今日使用数量已达上限")
                    raise ValueError(f"{normalized_model_id} 今日剩余 {remaining} 张，不足以生成 {normalized_amount} 张")
                limit["reserved"] = reserved + normalized_amount
                limit["date"] = str(limit.get("date") or "").strip() or _today_key()
                limit["updated_at"] = _now_iso()
                next_limits = [
                    limit if self._clean(entry.get("model_id")) == normalized_model_id else entry
                    for entry in limits
                ]
                next_item = dict(item)
                next_item[source_field] = next_limits
                next_items = list(items)
                next_items[index] = next_item
                result = self._public_item(next_item)
                result["_internal_image_model_daily_limit_reservation"] = {
                    "model_id": normalized_model_id,
                    "amount": normalized_amount,
                    "scope": reservation_scope,
                    "limit_scope": limit_scope,
                }
                return next_items, result
            return items, {"_internal_image_model_daily_limit_reservation": {"amount": 0}}

        return self._update_items_locked(mutate)

    def consume_internal_image_model_daily_limit(
        self,
        key_id: str,
        *,
        model_id: str,
        amount: int,
        scope: str = "",
    ) -> dict[str, object] | None:
        normalized_id = self._clean(key_id)
        normalized_model_id = self._clean(model_id)
        normalized_amount = max(0, int(amount or 0))
        if not normalized_id or not normalized_model_id or normalized_amount <= 0:
            return None
        normalized_scope = self._clean(scope)
        if normalized_scope == "default_total":
            return config.consume_internal_image_model_total_daily_limit(normalized_model_id, normalized_amount)
        source_field = "image_model_daily_default_usages" if normalized_scope == "default_user" else "image_model_daily_limits"

        def mutate(items: list[dict[str, object]]) -> tuple[list[dict[str, object]], dict[str, object] | None]:
            for index, item in enumerate(items):
                if item.get("id") != normalized_id or item.get("role") != "user":
                    continue
                limits = _normalize_image_model_daily_limits(item.get(source_field))
                changed = False
                for limit in limits:
                    if self._clean(limit.get("model_id")) != normalized_model_id:
                        continue
                    limit = _reset_image_model_daily_limit(limit)
                    reserved = _to_nonnegative_int(limit.get("reserved"))
                    consumed = min(reserved, normalized_amount)
                    limit["reserved"] = reserved - consumed
                    limit["used"] = _to_nonnegative_int(limit.get("used")) + consumed
                    limit["updated_at"] = _now_iso()
                    changed = True
                    next_limits = [
                        limit if self._clean(entry.get("model_id")) == normalized_model_id else entry
                        for entry in limits
                    ]
                    next_item = dict(item)
                    next_item[source_field] = next_limits
                    next_items = list(items)
                    next_items[index] = next_item
                    return next_items, self._public_item(next_item)
                if not changed:
                    return items, None
            return items, None

        return self._update_items_locked(mutate)

    def refund_internal_image_model_daily_limit(
        self,
        key_id: str,
        *,
        model_id: str,
        amount: int,
        scope: str = "",
    ) -> dict[str, object] | None:
        normalized_id = self._clean(key_id)
        normalized_model_id = self._clean(model_id)
        normalized_amount = max(0, int(amount or 0))
        if not normalized_id or not normalized_model_id or normalized_amount <= 0:
            return None
        normalized_scope = self._clean(scope)
        if normalized_scope == "default_total":
            return config.refund_internal_image_model_total_daily_limit(normalized_model_id, normalized_amount)
        source_field = "image_model_daily_default_usages" if normalized_scope == "default_user" else "image_model_daily_limits"

        def mutate(items: list[dict[str, object]]) -> tuple[list[dict[str, object]], dict[str, object] | None]:
            for index, item in enumerate(items):
                if item.get("id") != normalized_id or item.get("role") != "user":
                    continue
                limits = _normalize_image_model_daily_limits(item.get(source_field))
                for limit in limits:
                    if self._clean(limit.get("model_id")) != normalized_model_id:
                        continue
                    limit = _reset_image_model_daily_limit(limit)
                    reserved = _to_nonnegative_int(limit.get("reserved"))
                    refunded = min(reserved, normalized_amount)
                    limit["reserved"] = reserved - refunded
                    limit["updated_at"] = _now_iso()
                    next_limits = [
                        limit if self._clean(entry.get("model_id")) == normalized_model_id else entry
                        for entry in limits
                    ]
                    next_item = dict(item)
                    next_item[source_field] = next_limits
                    next_items = list(items)
                    next_items[index] = next_item
                    return next_items, self._public_item(next_item)
            return items, None

        return self._update_items_locked(mutate)

    def delete_key(self, key_id: str, *, role: AuthRole | None = None) -> bool:
        normalized_id = self._clean(key_id)
        if not normalized_id:
            return False

        # Database-backed stores can delete one row directly. Avoid loading,
        # normalizing, and rewriting every user record: a single malformed
        # legacy row should not prevent an unrelated user from being deleted.
        merge_auth_keys = getattr(self.storage, "merge_auth_keys", None)
        if callable(merge_auth_keys):
            with self._lock:
                items = self._load(fallback=self._items)
                target = next(
                    (
                        item
                        for item in items
                        if item.get("id") == normalized_id
                        and (role is None or item.get("role") == role)
                    ),
                    None,
                )
                if target is None:
                    self._items = items
                    return False
                merge_auth_keys([], [normalized_id])
                self._items = [item for item in items if item.get("id") != normalized_id]
                return True

        def mutate(items: list[dict[str, object]]) -> tuple[list[dict[str, object]], bool]:
            next_items = [
                item
                for item in items
                if not (item.get("id") == normalized_id and (role is None or item.get("role") == role))
            ]
            return next_items, len(next_items) < len(items)

        return self._update_items_locked(mutate)

    def batch_mutate_keys(
        self,
        key_ids: list[str],
        *,
        action: Literal["enable", "disable", "delete"],
        role: AuthRole,
    ) -> dict[str, object]:
        """Enable, disable, or delete multiple users in one storage transaction."""
        normalized_ids = list(
            dict.fromkeys(
                item_id
                for item_id in (self._clean(key_id) for key_id in key_ids)
                if item_id
            )
        )
        if not normalized_ids:
            raise ValueError("请至少选择一个用户")
        if action not in {"enable", "disable", "delete"}:
            raise ValueError("不支持的批量操作")
        if role not in {"user", "normal"}:
            raise ValueError("不支持的用户类型")

        selected_ids = set(normalized_ids)
        with self._lock:
            items = self._load(fallback=self._items)
            matched_ids = {
                str(item.get("id") or "")
                for item in items
                if item.get("id") in selected_ids and item.get("role") == role
            }

            if action == "delete":
                next_items = [item for item in items if str(item.get("id") or "") not in matched_ids]
                if matched_ids:
                    self._items = next_items
                    self._save(changed_items=[], deleted_ids=list(matched_ids))
                else:
                    self._items = items
                changed_count = len(matched_ids)
            else:
                enabled = action == "enable"
                changed_items: list[dict[str, object]] = []
                next_items: list[dict[str, object]] = []
                for item in items:
                    item_id = str(item.get("id") or "")
                    if item_id not in matched_ids or bool(item.get("enabled")) == enabled:
                        next_items.append(item)
                        continue
                    next_item = dict(item)
                    next_item["enabled"] = enabled
                    next_items.append(next_item)
                    changed_items.append(next_item)
                self._items = next_items
                if changed_items:
                    self._save(changed_items=changed_items, deleted_ids=[])
                changed_count = len(changed_items)

            return {
                "action": action,
                "requested": len(normalized_ids),
                "matched": len(matched_ids),
                "changed": changed_count,
            }

    def batch_update_image_permissions(
        self,
        updates: list[dict[str, object]],
        *,
        role: AuthRole,
    ) -> dict[str, object]:
        """Update image routing permissions for many users in one storage transaction."""
        if role not in {"user", "normal"}:
            raise ValueError("不支持的用户类型")
        normalized_updates = {
            self._clean(update.get("id")): update
            for update in updates
            if isinstance(update, dict) and self._clean(update.get("id"))
        }
        if not normalized_updates:
            raise ValueError("请至少选择一个用户")

        with self._lock:
            items = self._load(fallback=self._items)
            changed_items: list[dict[str, object]] = []
            next_items: list[dict[str, object]] = []
            matched = 0
            for item in items:
                item_id = self._clean(item.get("id"))
                update = normalized_updates.get(item_id)
                if update is None or item.get("role") != role:
                    next_items.append(item)
                    continue
                matched += 1
                next_item = dict(item)
                if "third_party_image_enabled" in update:
                    next_item["third_party_image_enabled"] = bool(update.get("third_party_image_enabled"))
                if "third_party_image_group_ids" in update:
                    next_item["third_party_image_group_ids"] = _normalize_string_list(update.get("third_party_image_group_ids"))
                if "third_party_image_model_ids" in update:
                    next_item["third_party_image_model_ids"] = _normalize_string_list(update.get("third_party_image_model_ids"))
                next_items.append(next_item)
                if next_item != item:
                    changed_items.append(next_item)

            self._items = next_items
            if changed_items:
                self._save(changed_items=changed_items, deleted_ids=[])
            return {
                "requested": len(normalized_updates),
                "matched": matched,
                "changed": len(changed_items),
            }

    def credit_summary(self, identity: dict[str, object]) -> dict[str, object]:
        role = self._clean(identity.get("role")).lower()
        if role != "normal":
            return {
                "role": role,
                "metered": False,
                "credits": None,
                "paid_credits": None,
                "reserved_credits": 0,
                "used_credits": None,
                "total_recharged": None,
                "permanent_credits": None,
                "expiring_credits": 0,
                "expiring_reserved_credits": 0,
                "credit_grants": [],
                "package_credits": 0,
                "package_grants": [],
                "daily_credits": 0,
                "daily_credits_remaining": 0,
                "daily_credits_used": None,
                "daily_bonus_credits": 0,
                "daily_reserved_credits": 0,
                "daily_credits_date": "",
                "checkin_credits": 0,
                "checkin_credits_expires_at": "",
                "checkin_credit_batches": [],
                "third_party_image_enabled": False,
                "third_party_image_wallets": [],
            }
        item = self.get_key(self._clean(identity.get("id")), role="normal")
        if item is None:
            return {
                "role": role,
                "metered": True,
                "credits": 0,
                "paid_credits": 0,
                "reserved_credits": 0,
                "used_credits": 0,
                "total_recharged": 0,
                "permanent_credits": 0,
                "expiring_credits": 0,
                "expiring_reserved_credits": 0,
                "credit_grants": [],
                "package_credits": 0,
                "package_grants": [],
                "daily_credits": config.normal_user_daily_credits,
                "daily_credits_remaining": 0,
                "daily_credits_used": 0,
                "daily_bonus_credits": 0,
                "daily_reserved_credits": 0,
                "daily_credits_date": _today_key(),
                "checkin_credits": 0,
                "checkin_credits_expires_at": "",
                "checkin_credit_batches": [],
                "third_party_image_enabled": False,
                "third_party_image_wallets": [],
            }
        return self.credit_summary_from_public_item(item)

    def credit_summary_from_public_item(self, item: dict[str, object]) -> dict[str, object]:
        return {
            "role": "normal",
            "metered": True,
            "credits": _credit_number(item.get("credits")),
            "paid_credits": _credit_number(item.get("paid_credits")),
            "reserved_credits": _credit_number(item.get("reserved_credits")),
            "used_credits": _credit_number(item.get("used_credits")),
            "total_recharged": _credit_number(item.get("total_recharged")),
            "permanent_credits": _credit_number(item.get("permanent_credits")),
            "expiring_credits": _credit_number(item.get("expiring_credits")),
            "expiring_reserved_credits": _credit_number(item.get("expiring_reserved_credits")),
            "credit_grants": item.get("credit_grants") if isinstance(item.get("credit_grants"), list) else [],
            "package_credits": _credit_number(item.get("package_credits")),
            "package_grants": item.get("package_grants") if isinstance(item.get("package_grants"), list) else [],
            "daily_credits": _credit_number(item.get("daily_credits")),
            "daily_credits_remaining": _credit_number(item.get("daily_credits_remaining")),
            "daily_credits_used": _credit_number(item.get("daily_credits_used")),
            "daily_bonus_credits": _credit_number(item.get("daily_bonus_credits")),
            "daily_reserved_credits": _credit_number(item.get("daily_reserved_credits")),
            "daily_credits_date": self._clean(item.get("daily_credits_date")),
            "checkin_credits": _credit_number(item.get("checkin_credits")),
            "checkin_credits_expires_at": self._clean(item.get("checkin_credits_expires_at")),
            "checkin_credit_batches": item.get("checkin_credit_batches") if isinstance(item.get("checkin_credit_batches"), list) else [],
            "third_party_image_enabled": bool(item.get("third_party_image_enabled")),
            "third_party_image_wallets": item.get("third_party_image_wallets") if isinstance(item.get("third_party_image_wallets"), list) else [],
        }

    def _adjust_credits_in_items(
        self,
        items: list[dict[str, object]],
        key_id: str,
        delta: int,
        *,
        note: str = "",
        expires_at: str | None = None,
        source: str = "",
        code_id: str = "",
    ) -> tuple[list[dict[str, object]], dict[str, object]]:
        amount = int(delta)
        for index, item in enumerate(items):
            if item.get("id") != key_id or item.get("role") != "normal":
                continue
            next_item = dict(item)
            credits = _credit_decimal(item.get("credits"))
            if amount > 0 and _is_future_datetime(expires_at):
                grants = _normalize_credit_grants(item.get("credit_grants"))
                grants.append(
                    {
                        "id": uuid.uuid4().hex[:12],
                        "remaining": amount,
                        "reserved": 0,
                        "expires_at": str(expires_at or "").strip(),
                        "created_at": _now_iso(),
                        "source": self._clean(source or note),
                        "code_id": self._clean(code_id),
                    }
                )
                grants.sort(key=lambda grant: str(grant.get("expires_at") or ""))
                next_item["credit_grants"] = grants
                next_item["credits"] = _credit_number(credits)
            else:
                next_credits = credits + Decimal(amount)
                if next_credits < 0:
                    raise ValueError("用户剩余次数不足，不能扣成负数")
                next_item["credits"] = _credit_number(next_credits)
                # 签到账本：只有落进基础积分池的变动才记账。
                # 发放（source=checkin）累加；手动扣分按「签到积分最先被消耗」扣减。
                checkin_remaining = _credit_decimal(item.get("checkin_credit_remaining"))
                if amount > 0:
                    if self._clean(source) == "checkin":
                        checkin_remaining += Decimal(amount)
                elif amount < 0:
                    checkin_remaining = max(Decimal("0"), checkin_remaining + Decimal(amount))
                next_item["checkin_credit_remaining"] = _credit_number(checkin_remaining)
            if amount > 0:
                next_item["total_recharged"] = _credit_number(_credit_decimal(item.get("total_recharged")) + Decimal(amount))
            next_items = list(items)
            next_items[index] = next_item
            return next_items, self._public_item(next_item)
        raise ValueError("普通用户不存在")

    def adjust_credits(self, key_id: str, delta: int, *, note: str = "", expires_at: str | None = None, source: str = "", code_id: str = "") -> dict[str, object]:
        normalized_id = self._clean(key_id)
        if not normalized_id:
            raise ValueError("用户不存在")

        def mutate(items: list[dict[str, object]]) -> tuple[list[dict[str, object]], dict[str, object]]:
            return self._adjust_credits_in_items(
                items,
                normalized_id,
                int(delta),
                note=note,
                expires_at=expires_at,
                source=source,
                code_id=code_id,
            )

        return self._update_items_locked(mutate)

    def _add_third_party_image_wallet_in_items(
        self,
        items: list[dict[str, object]],
        key_id: str,
        *,
        group_id: str,
        model_id: str,
        amount: int,
        auto_enable: bool = True,
    ) -> tuple[list[dict[str, object]], dict[str, object]]:
        normalized_id = self._clean(key_id)
        normalized_group_id = self._clean(group_id)
        normalized_model_id = self._clean(model_id)
        normalized_amount = max(1, _to_nonnegative_int(amount))
        if not normalized_group_id:
            raise ValueError("image API group is required")
        if not normalized_model_id:
            raise ValueError("image API model is required")
        for index, item in enumerate(items):
            if item.get("id") != normalized_id or item.get("role") != "normal":
                continue
            wallets = _normalize_third_party_image_wallets(item.get("third_party_image_wallets"))
            wallet_key = f"{normalized_group_id}:{normalized_model_id}"
            wallet = next((entry for entry in wallets if self._clean(entry.get("id")) == wallet_key), None)
            if wallet is None:
                wallet = {
                    "id": wallet_key,
                    "group_id": normalized_group_id,
                    "model_id": normalized_model_id,
                    "remaining": 0,
                    "reserved": 0,
                    "used": 0,
                    "total_recharged": 0,
                    "updated_at": "",
                }
                wallets.append(wallet)
            wallet["remaining"] = _credit_number(_credit_decimal(wallet.get("remaining")) + _credit_decimal(normalized_amount))
            wallet["total_recharged"] = _credit_number(_credit_decimal(wallet.get("total_recharged")) + _credit_decimal(normalized_amount))
            wallet["updated_at"] = _now_iso()
            next_item = dict(item)
            next_item["third_party_image_wallets"] = wallets
            if auto_enable:
                next_item["third_party_image_enabled"] = True
                if normalized_model_id == THIRD_PARTY_IMAGE_GROUP_WALLET_MODEL_ID:
                    group_ids = set(_normalize_string_list(next_item.get("third_party_image_group_ids")))
                    group_ids.add(normalized_group_id)
                    next_item["third_party_image_group_ids"] = sorted(group_ids)
                    next_item["third_party_image_model_ids"] = []
                else:
                    model_ids = set(_normalize_string_list(next_item.get("third_party_image_model_ids")))
                    model_ids.add(f"{normalized_group_id}:{normalized_model_id}")
                    next_item["third_party_image_model_ids"] = sorted(model_ids)
            next_items = list(items)
            next_items[index] = next_item
            return next_items, self._public_item(next_item)
        raise ValueError("normal user not found")

    def adjust_third_party_image_wallet(
        self,
        key_id: str,
        *,
        group_id: str,
        model_id: str,
        delta: int,
        source: str = "",
        code_id: str = "",
        auto_enable: bool = True,
    ) -> dict[str, object]:
        normalized_id = self._clean(key_id)
        normalized_group_id = self._clean(group_id)
        normalized_model_id = self._clean(model_id)
        amount = int(delta)
        if not normalized_id:
            raise ValueError("user not found")
        if not normalized_group_id:
            raise ValueError("image API group is required")
        if not normalized_model_id:
            raise ValueError("image API model is required")

        def mutate(items: list[dict[str, object]]) -> tuple[list[dict[str, object]], dict[str, object]]:
            for index, item in enumerate(items):
                if item.get("id") != normalized_id or item.get("role") != "normal":
                    continue
                wallets = _normalize_third_party_image_wallets(item.get("third_party_image_wallets"))
                wallet_key = f"{normalized_group_id}:{normalized_model_id}"
                wallet = next((entry for entry in wallets if self._clean(entry.get("id")) == wallet_key), None)
                if wallet is None:
                    wallet = {
                        "id": wallet_key,
                        "group_id": normalized_group_id,
                        "model_id": normalized_model_id,
                        "remaining": 0,
                        "reserved": 0,
                        "used": 0,
                        "total_recharged": 0,
                        "updated_at": "",
                    }
                    wallets.append(wallet)
                next_remaining = _credit_decimal(wallet.get("remaining")) + Decimal(amount)
                if next_remaining < Decimal("0"):
                    raise ValueError("4K image balance is not enough")
                wallet["remaining"] = _credit_number(next_remaining)
                wallet["updated_at"] = _now_iso()
                if amount > 0:
                    wallet["total_recharged"] = _credit_number(_credit_decimal(wallet.get("total_recharged")) + Decimal(amount))
                next_item = dict(item)
                next_item["third_party_image_wallets"] = wallets
                if auto_enable:
                    next_item["third_party_image_enabled"] = True
                    if normalized_model_id == THIRD_PARTY_IMAGE_GROUP_WALLET_MODEL_ID:
                        group_ids = set(_normalize_string_list(next_item.get("third_party_image_group_ids")))
                        group_ids.add(normalized_group_id)
                        next_item["third_party_image_group_ids"] = sorted(group_ids)
                        next_item["third_party_image_model_ids"] = []
                    else:
                        model_ids = set(_normalize_string_list(next_item.get("third_party_image_model_ids")))
                        model_ids.add(f"{normalized_group_id}:{normalized_model_id}")
                        next_item["third_party_image_model_ids"] = sorted(model_ids)
                next_items = list(items)
                next_items[index] = next_item
                return next_items, self._public_item(next_item)
            raise ValueError("normal user not found")

        return self._update_items_locked(mutate)

    def reserve_third_party_image_wallet(
        self,
        key_id: str,
        *,
        group_id: str,
        model_id: str,
        amount: int | float,
    ) -> dict[str, object]:
        normalized_id = self._clean(key_id)
        normalized_group_id = self._clean(group_id)
        normalized_model_id = self._clean(model_id)
        normalized_amount = _credit_decimal(amount)
        if normalized_amount <= 0:
            raise ValueError("image API model balance amount must be positive")

        def mutate(items: list[dict[str, object]]) -> tuple[list[dict[str, object]], dict[str, object]]:
            for index, item in enumerate(items):
                if item.get("id") != normalized_id or item.get("role") != "normal":
                    continue
                wallets = _normalize_third_party_image_wallets(item.get("third_party_image_wallets"))
                wallet = _find_third_party_wallet_for_reserve(wallets, normalized_group_id, normalized_model_id, normalized_amount)
                if wallet is None:
                    raise ValueError("image API model balance is not enough")
                remaining = _credit_decimal(wallet.get("remaining"))
                if remaining < normalized_amount:
                    raise ValueError("image API model balance is not enough")
                wallet["remaining"] = _credit_number(remaining - normalized_amount)
                wallet["reserved"] = _credit_number(_credit_decimal(wallet.get("reserved")) + normalized_amount)
                wallet["updated_at"] = _now_iso()
                next_item = dict(item)
                next_item["third_party_image_wallets"] = wallets
                next_items = list(items)
                next_items[index] = next_item
                result = self._public_item(next_item)
                result["_third_party_image_reservation"] = {
                    "group_id": normalized_group_id,
                    "model_id": normalized_model_id,
                    "amount": _credit_number(normalized_amount),
                }
                return next_items, result
            raise ValueError("normal user not found")

        return self._update_items_locked(mutate)

    def consume_third_party_image_wallet(
        self,
        key_id: str,
        *,
        group_id: str,
        model_id: str,
        amount: int | float,
    ) -> dict[str, object] | None:
        normalized_id = self._clean(key_id)
        normalized_group_id = self._clean(group_id)
        normalized_model_id = self._clean(model_id)
        normalized_amount = _credit_decimal(amount)
        if normalized_amount <= 0:
            return None

        def mutate(items: list[dict[str, object]]) -> tuple[list[dict[str, object]], dict[str, object] | None]:
            for index, item in enumerate(items):
                if item.get("id") != normalized_id or item.get("role") != "normal":
                    continue
                wallets = _normalize_third_party_image_wallets(item.get("third_party_image_wallets"))
                wallet = _find_third_party_wallet_for_settle(wallets, normalized_group_id, normalized_model_id)
                if wallet is not None:
                    reserved = _credit_decimal(wallet.get("reserved"))
                    consumed = min(reserved, normalized_amount)
                    wallet["reserved"] = _credit_number(reserved - consumed)
                    wallet["used"] = _credit_number(_credit_decimal(wallet.get("used")) + consumed)
                    wallet["updated_at"] = _now_iso()
                    next_item = dict(item)
                    next_item["third_party_image_wallets"] = wallets
                    next_items = list(items)
                    next_items[index] = next_item
                    return next_items, self._public_item(next_item)
                return items, None
            return items, None

        return self._update_items_locked(mutate)

    def refund_third_party_image_wallet(
        self,
        key_id: str,
        *,
        group_id: str,
        model_id: str,
        amount: int | float,
    ) -> dict[str, object] | None:
        normalized_id = self._clean(key_id)
        normalized_group_id = self._clean(group_id)
        normalized_model_id = self._clean(model_id)
        normalized_amount = _credit_decimal(amount)
        if normalized_amount <= 0:
            return None

        def mutate(items: list[dict[str, object]]) -> tuple[list[dict[str, object]], dict[str, object] | None]:
            for index, item in enumerate(items):
                if item.get("id") != normalized_id or item.get("role") != "normal":
                    continue
                wallets = _normalize_third_party_image_wallets(item.get("third_party_image_wallets"))
                wallet = _find_third_party_wallet_for_settle(wallets, normalized_group_id, normalized_model_id)
                if wallet is not None:
                    reserved = _credit_decimal(wallet.get("reserved"))
                    refunded = min(reserved, normalized_amount)
                    wallet["reserved"] = _credit_number(reserved - refunded)
                    wallet["remaining"] = _credit_number(_credit_decimal(wallet.get("remaining")) + refunded)
                    wallet["updated_at"] = _now_iso()
                    next_item = dict(item)
                    next_item["third_party_image_wallets"] = wallets
                    next_items = list(items)
                    next_items[index] = next_item
                    return next_items, self._public_item(next_item)
                return items, None
            return items, None

        return self._update_items_locked(mutate)

    def _add_daily_credits_in_items(
        self,
        items: list[dict[str, object]],
        key_id: str,
        amount: int,
        *,
        source: str = "",
        code_id: str = "",
        model_ids: list[str] | None = None,
    ) -> tuple[list[dict[str, object]], dict[str, object]]:
        normalized_amount = max(1, _to_nonnegative_int(amount))
        today = _today_key()
        for index, item in enumerate(items):
            if item.get("id") != key_id or item.get("role") != "normal":
                continue
            normalized_item = self._reset_daily_if_needed(dict(item))
            next_item = dict(normalized_item)
            next_item["daily_credits_date"] = today
            next_item["daily_bonus_credits"] = _to_nonnegative_int(normalized_item.get("daily_bonus_credits")) + normalized_amount
            normalized_model_ids = _normalize_string_list(model_ids) or [DAILY_GIFT_ACCOUNT_POOL_MODEL_ID]
            merged_model_ids = set(self._daily_bonus_model_ids(normalized_item))
            merged_model_ids.update(normalized_model_ids)
            next_item["daily_bonus_model_ids"] = sorted(merged_model_ids)
            third_party_model_ids = {
                model_id
                for model_id in normalized_model_ids
                if model_id != DAILY_GIFT_ACCOUNT_POOL_MODEL_ID and ":" in model_id
            }
            if third_party_model_ids:
                next_item["third_party_image_enabled"] = True
                permitted_model_ids = set(_normalize_string_list(next_item.get("third_party_image_model_ids")))
                permitted_model_ids.update(third_party_model_ids)
                next_item["third_party_image_model_ids"] = sorted(permitted_model_ids)
            next_items = list(items)
            next_items[index] = next_item
            return next_items, self._public_item(next_item)
        raise ValueError("普通用户不存在")

    def add_daily_credits(
        self,
        key_id: str,
        amount: int,
        *,
        source: str = "",
        code_id: str = "",
        model_ids: list[str] | None = None,
    ) -> dict[str, object]:
        normalized_id = self._clean(key_id)
        if not normalized_id:
            raise ValueError("用户不存在")

        def mutate(items: list[dict[str, object]]) -> tuple[list[dict[str, object]], dict[str, object]]:
            return self._add_daily_credits_in_items(
                items,
                normalized_id,
                int(amount),
                source=source,
                code_id=code_id,
                model_ids=model_ids,
            )

        return self._update_items_locked(mutate)

    def _add_package_grant_in_items(
        self,
        items: list[dict[str, object]],
        key_id: str,
        package_plan: dict[str, object],
        *,
        source: str = "",
        code_id: str = "",
    ) -> tuple[list[dict[str, object]], dict[str, object]]:
        duration_days = max(1, _to_nonnegative_int(package_plan.get("duration_days")) or 1)
        now = datetime.now(timezone.utc)
        grant = {
            "id": uuid.uuid4().hex[:12],
            "name": self._clean(package_plan.get("name")) or "套餐",
            "created_at": now.isoformat(),
            "starts_at": now.isoformat(),
            "expires_at": (now + timedelta(days=duration_days)).isoformat(),
            "source": self._clean(source),
            "code_id": self._clean(code_id),
            "daily_limit": _to_nonnegative_int(package_plan.get("daily_limit")),
            "weekly_limit": _to_nonnegative_int(package_plan.get("weekly_limit")),
            "monthly_limit": _to_nonnegative_int(package_plan.get("monthly_limit")),
            "total_limit": _to_nonnegative_int(package_plan.get("total_limit")),
            "used_total": 0,
            "reserved_total": 0,
            "used_daily": 0,
            "reserved_daily": 0,
            "used_weekly": 0,
            "reserved_weekly": 0,
            "used_monthly": 0,
            "reserved_monthly": 0,
        }
        if max(
            _to_nonnegative_int(grant.get("daily_limit")),
            _to_nonnegative_int(grant.get("weekly_limit")),
            _to_nonnegative_int(grant.get("monthly_limit")),
            _to_nonnegative_int(grant.get("total_limit")),
        ) <= 0:
            raise ValueError("套餐卡密至少需要设置一个日/周/月/总限额")
        grant = _reset_package_periods(grant)
        for index, item in enumerate(items):
            if item.get("id") != key_id or item.get("role") != "normal":
                continue
            normalized_item = self._reset_daily_if_needed(dict(item))
            grants = _active_package_grants(normalized_item)
            grants.append(grant)
            next_item = dict(normalized_item)
            next_item["package_grants"] = grants
            next_items = list(items)
            next_items[index] = next_item
            return next_items, self._public_item(next_item)
        raise ValueError("普通用户不存在")

    def add_package_grant(self, key_id: str, package_plan: dict[str, object], *, source: str = "", code_id: str = "") -> dict[str, object]:
        normalized_id = self._clean(key_id)
        if not normalized_id:
            raise ValueError("用户不存在")

        def mutate(items: list[dict[str, object]]) -> tuple[list[dict[str, object]], dict[str, object]]:
            return self._add_package_grant_in_items(
                items,
                normalized_id,
                package_plan,
                source=source,
                code_id=code_id,
            )

        return self._update_items_locked(mutate)

    def reserve_credits(self, key_id: str, amount: int | float, *, model_id: str = "") -> dict[str, object]:
        """按固定优先级从各积分池预扣，并在返回值里带上 `_credit_reservation` 明细。
        扣费顺序（原则：先到期的先用）：
          1. 每日免费额度 `daily_credits_remaining` —— 当天有效，次日清零，最先扣
          2. 套餐券 `package_grants` —— 受日 / 周 / 月限额约束
          3. 限时券 `credit_grants` —— 带 `expires_at`，先到期的先扣
          4. 基础积分池 `credits` —— 永久有效，最后扣
        签到赠送的积分（`perform_checkin` 发放、无 `expires_at`）落在第 4 项基础积分池，
        即在整个链条上排最后；但在**该池内部**它最先被消耗（赠送优先消费，
        见 `consume_reserved_credits` 里的签到账本）。各池的消耗归属见
        `test/test_checkin_credit_ledger.py`。
        """
        amount_decimal = _credit_decimal(amount)
        if amount_decimal <= 0:
            result = self.credit_summary({"id": key_id, "role": "normal"})
            result["_credit_reservation"] = {"daily": 0, "packages": [], "paid": 0, "permanent_paid": 0, "grants": []}
            return result
        normalized_id = self._clean(key_id)

        def mutate(items: list[dict[str, object]]) -> tuple[list[dict[str, object]], dict[str, object]]:
            for index, item in enumerate(items):
                if item.get("id") != normalized_id or item.get("role") != "normal":
                    continue
                normalized_item = self._reset_daily_if_needed(dict(item))
                credits = _credit_decimal(normalized_item.get("credits"))
                grants = _normalize_credit_grants(normalized_item.get("credit_grants"))
                package_grants = _active_package_grants(normalized_item)
                expiring_available = sum((_credit_decimal(grant.get("remaining")) for grant in grants if _is_future_datetime(grant.get("expires_at"))), Decimal("0"))
                daily_available = _credit_decimal(self._daily_remaining(normalized_item, model_id))
                daily_to_reserve = min(amount_decimal, daily_available)
                remaining_to_reserve = amount_decimal - daily_to_reserve
                package_reservations: list[dict[str, object]] = []
                for grant in package_grants:
                    if remaining_to_reserve <= 0:
                        break
                    take = min(_credit_decimal(_package_available(grant)), remaining_to_reserve)
                    if take <= 0:
                        continue
                    if _to_nonnegative_int(grant.get("total_limit")) > 0:
                        grant["reserved_total"] = _credit_number(_credit_decimal(grant.get("reserved_total")) + take)
                    if _to_nonnegative_int(grant.get("daily_limit")) > 0:
                        grant["reserved_daily"] = _credit_number(_credit_decimal(grant.get("reserved_daily")) + take)
                    if _to_nonnegative_int(grant.get("weekly_limit")) > 0:
                        grant["reserved_weekly"] = _credit_number(_credit_decimal(grant.get("reserved_weekly")) + take)
                    if _to_nonnegative_int(grant.get("monthly_limit")) > 0:
                        grant["reserved_monthly"] = _credit_number(_credit_decimal(grant.get("reserved_monthly")) + take)
                    package_reservations.append({"id": grant.get("id"), "amount": _credit_number(take)})
                    remaining_to_reserve -= take
                paid_to_reserve = remaining_to_reserve
                if credits + expiring_available < paid_to_reserve:
                    raise ValueError("剩余积分不足，请先充值")
                next_item = dict(normalized_item)
                remaining_paid_to_reserve = paid_to_reserve
                grant_reservations: list[dict[str, object]] = []
                for grant in grants:
                    if remaining_paid_to_reserve <= 0 or not _is_future_datetime(grant.get("expires_at")):
                        continue
                    grant_remaining = _credit_decimal(grant.get("remaining"))
                    take = min(grant_remaining, remaining_paid_to_reserve)
                    if take <= 0:
                        continue
                    grant["remaining"] = _credit_number(grant_remaining - take)
                    grant["reserved"] = _credit_number(_credit_decimal(grant.get("reserved")) + take)
                    grant_reservations.append({"id": grant.get("id"), "amount": _credit_number(take)})
                    remaining_paid_to_reserve -= take
                permanent_to_reserve = remaining_paid_to_reserve
                next_item["credits"] = _credit_number(credits - permanent_to_reserve)
                next_item["reserved_credits"] = _credit_number(_credit_decimal(normalized_item.get("reserved_credits")) + permanent_to_reserve)
                next_item["daily_reserved_credits"] = _credit_number(_credit_decimal(normalized_item.get("daily_reserved_credits")) + daily_to_reserve)
                next_item["credit_grants"] = grants
                next_item["package_grants"] = package_grants
                result = self._public_item(next_item)
                result["_credit_reservation"] = {
                    "daily": _credit_number(daily_to_reserve),
                    "packages": package_reservations,
                    "paid": _credit_number(paid_to_reserve),
                    "permanent_paid": _credit_number(permanent_to_reserve),
                    "grants": grant_reservations,
                }
                next_items = list(items)
                next_items[index] = next_item
                return next_items, result
            raise ValueError("普通用户不存在")

        return self._update_items_locked(mutate)

    def consume_reserved_credits(
        self,
        key_id: str,
        amount: int | float,
        *,
        daily_amount: int | float | None = None,
        package_reservations: list[dict[str, object]] | None = None,
        paid_amount: int | float | None = None,
        grant_reservations: list[dict[str, object]] | None = None,
    ) -> dict[str, object] | None:
        amount_decimal = _credit_decimal(amount)
        if amount_decimal <= 0:
            return self.get_key(key_id, role="normal")
        normalized_id = self._clean(key_id)

        def mutate(items: list[dict[str, object]]) -> tuple[list[dict[str, object]], dict[str, object] | None]:
            for index, item in enumerate(items):
                if item.get("id") != normalized_id or item.get("role") != "normal":
                    continue
                normalized_item = self._reset_daily_if_needed(dict(item))
                daily_reserved = _credit_decimal(normalized_item.get("daily_reserved_credits"))
                paid_reserved = _credit_decimal(normalized_item.get("reserved_credits"))
                grants = _normalize_credit_grants(normalized_item.get("credit_grants"))
                package_grants = _settleable_package_grants(normalized_item)
                package_consumed = Decimal("0")
                for reservation in package_reservations or []:
                    if not isinstance(reservation, dict):
                        continue
                    package_id = self._clean(reservation.get("id"))
                    requested = _credit_decimal(reservation.get("amount"))
                    if not package_id or requested <= 0:
                        continue
                    for grant in package_grants:
                        if self._clean(grant.get("id")) != package_id:
                            continue
                        consumed_from_package = Decimal("0")
                        if _to_nonnegative_int(grant.get("total_limit")) > 0:
                            take = min(requested, _credit_decimal(grant.get("reserved_total")))
                            grant["reserved_total"] = _credit_number(_credit_decimal(grant.get("reserved_total")) - take)
                            grant["used_total"] = _credit_number(_credit_decimal(grant.get("used_total")) + take)
                            consumed_from_package = max(consumed_from_package, take)
                        if _to_nonnegative_int(grant.get("daily_limit")) > 0:
                            take = min(requested, _credit_decimal(grant.get("reserved_daily")))
                            grant["reserved_daily"] = _credit_number(_credit_decimal(grant.get("reserved_daily")) - take)
                            grant["used_daily"] = _credit_number(_credit_decimal(grant.get("used_daily")) + take)
                            consumed_from_package = max(consumed_from_package, take)
                        if _to_nonnegative_int(grant.get("weekly_limit")) > 0:
                            take = min(requested, _credit_decimal(grant.get("reserved_weekly")))
                            grant["reserved_weekly"] = _credit_number(_credit_decimal(grant.get("reserved_weekly")) - take)
                            grant["used_weekly"] = _credit_number(_credit_decimal(grant.get("used_weekly")) + take)
                            consumed_from_package = max(consumed_from_package, take)
                        if _to_nonnegative_int(grant.get("monthly_limit")) > 0:
                            take = min(requested, _credit_decimal(grant.get("reserved_monthly")))
                            grant["reserved_monthly"] = _credit_number(_credit_decimal(grant.get("reserved_monthly")) - take)
                            grant["used_monthly"] = _credit_number(_credit_decimal(grant.get("used_monthly")) + take)
                            consumed_from_package = max(consumed_from_package, take)
                        package_consumed += consumed_from_package
                        break
                grant_consumed = Decimal("0")
                for reservation in grant_reservations or []:
                    if not isinstance(reservation, dict):
                        continue
                    grant_id = self._clean(reservation.get("id"))
                    requested = _credit_decimal(reservation.get("amount"))
                    if not grant_id or requested <= 0:
                        continue
                    for grant in grants:
                        if self._clean(grant.get("id")) != grant_id:
                            continue
                        consumed_from_grant = min(requested, _credit_decimal(grant.get("reserved")))
                        grant["reserved"] = _credit_number(_credit_decimal(grant.get("reserved")) - consumed_from_grant)
                        grant_consumed += consumed_from_grant
                        break
                if daily_amount is None and paid_amount is None:
                    daily_consumed = min(amount_decimal, daily_reserved)
                    paid_consumed = min(amount_decimal - daily_consumed, paid_reserved)
                else:
                    daily_consumed = min(_credit_decimal(daily_amount), daily_reserved)
                    paid_consumed = min(_credit_decimal(paid_amount), paid_reserved)
                consumed = daily_consumed + paid_consumed
                next_item = dict(normalized_item)
                # 签到账本按「签到积分最先被消耗」扣减；只有从**基础积分池**扣掉的部分算数
                # （每日免费额度 / 限时券 / 套餐券各自独立，不动签到账本）。
                checkin_remaining = _credit_decimal(normalized_item.get("checkin_credit_remaining"))
                if checkin_remaining > 0 and paid_consumed > 0:
                    checkin_remaining = max(Decimal("0"), checkin_remaining - paid_consumed)
                next_item["checkin_credit_remaining"] = _credit_number(checkin_remaining)
                next_item["daily_reserved_credits"] = _credit_number(daily_reserved - daily_consumed)
                next_item["daily_credits_used"] = _credit_number(_credit_decimal(normalized_item.get("daily_credits_used")) + daily_consumed)
                next_item["reserved_credits"] = _credit_number(paid_reserved - paid_consumed)
                next_item["credit_grants"] = grants
                next_item["package_grants"] = package_grants
                next_item["used_credits"] = _credit_number(_credit_decimal(normalized_item.get("used_credits")) + consumed + grant_consumed + package_consumed)
                next_items = list(items)
                next_items[index] = next_item
                return next_items, self._public_item(next_item)
            return items, None

        return self._update_items_locked(mutate)

    def refund_reserved_credits(
        self,
        key_id: str,
        amount: int | float,
        *,
        daily_amount: int | float | None = None,
        paid_amount: int | float | None = None,
        package_reservations: list[dict[str, object]] | None = None,
        grant_reservations: list[dict[str, object]] | None = None,
    ) -> dict[str, object] | None:
        amount_decimal = _credit_decimal(amount)
        if amount_decimal <= 0:
            return self.get_key(key_id, role="normal")
        normalized_id = self._clean(key_id)

        def mutate(items: list[dict[str, object]]) -> tuple[list[dict[str, object]], dict[str, object] | None]:
            for index, item in enumerate(items):
                if item.get("id") != normalized_id or item.get("role") != "normal":
                    continue
                normalized_item = self._reset_daily_if_needed(dict(item))
                daily_reserved = _credit_decimal(normalized_item.get("daily_reserved_credits"))
                paid_reserved = _credit_decimal(normalized_item.get("reserved_credits"))
                grants = _normalize_credit_grants(normalized_item.get("credit_grants"))
                package_grants = _settleable_package_grants(normalized_item)
                for reservation in package_reservations or []:
                    if not isinstance(reservation, dict):
                        continue
                    package_id = self._clean(reservation.get("id"))
                    requested = _credit_decimal(reservation.get("amount"))
                    if not package_id or requested <= 0:
                        continue
                    for grant in package_grants:
                        if self._clean(grant.get("id")) != package_id:
                            continue
                        if _to_nonnegative_int(grant.get("total_limit")) > 0:
                            take = min(requested, _credit_decimal(grant.get("reserved_total")))
                            grant["reserved_total"] = _credit_number(_credit_decimal(grant.get("reserved_total")) - take)
                        if _to_nonnegative_int(grant.get("daily_limit")) > 0:
                            take = min(requested, _credit_decimal(grant.get("reserved_daily")))
                            grant["reserved_daily"] = _credit_number(_credit_decimal(grant.get("reserved_daily")) - take)
                        if _to_nonnegative_int(grant.get("weekly_limit")) > 0:
                            take = min(requested, _credit_decimal(grant.get("reserved_weekly")))
                            grant["reserved_weekly"] = _credit_number(_credit_decimal(grant.get("reserved_weekly")) - take)
                        if _to_nonnegative_int(grant.get("monthly_limit")) > 0:
                            take = min(requested, _credit_decimal(grant.get("reserved_monthly")))
                            grant["reserved_monthly"] = _credit_number(_credit_decimal(grant.get("reserved_monthly")) - take)
                        break
                for reservation in grant_reservations or []:
                    if not isinstance(reservation, dict):
                        continue
                    grant_id = self._clean(reservation.get("id"))
                    requested = _credit_decimal(reservation.get("amount"))
                    if not grant_id or requested <= 0:
                        continue
                    for grant in grants:
                        if self._clean(grant.get("id")) != grant_id:
                            continue
                        refunded_from_grant = min(requested, _credit_decimal(grant.get("reserved")))
                        grant["reserved"] = _credit_number(_credit_decimal(grant.get("reserved")) - refunded_from_grant)
                        grant["remaining"] = _credit_number(_credit_decimal(grant.get("remaining")) + refunded_from_grant)
                        break
                if daily_amount is None and paid_amount is None:
                    daily_refunded = min(amount_decimal, daily_reserved)
                    paid_refunded = min(amount_decimal - daily_refunded, paid_reserved)
                else:
                    daily_refunded = min(_credit_decimal(daily_amount), daily_reserved)
                    paid_refunded = min(_credit_decimal(paid_amount), paid_reserved)
                next_item = dict(normalized_item)
                next_item["daily_reserved_credits"] = _credit_number(daily_reserved - daily_refunded)
                next_item["reserved_credits"] = _credit_number(paid_reserved - paid_refunded)
                next_item["credits"] = _credit_number(_credit_decimal(normalized_item.get("credits")) + paid_refunded)
                next_item["credit_grants"] = grants
                next_item["package_grants"] = package_grants
                next_items = list(items)
                next_items[index] = next_item
                return next_items, self._public_item(next_item)
            return items, None

        return self._update_items_locked(mutate)

    def authenticate(self, raw_key: str, *, touch: bool = True) -> dict[str, object] | None:
        candidate = self._clean(raw_key)
        if not candidate:
            return None
        candidate_hash = _hash_key(candidate)

        def find_match(items: list[dict[str, object]]) -> tuple[int, dict[str, object]] | None:
            for index, item in enumerate(items):
                if not bool(item.get("enabled", True)):
                    continue
                stored_hash = self._clean(item.get("key_hash"))
                if stored_hash and hmac.compare_digest(stored_hash, candidate_hash):
                    return index, item
            return None

        # ⚠️⚠️ 这一段跑在**事件循环线程**上（async 路由里同步调 require_identity），
        # 所以里面**绝不允许出现任何存储 IO**。锁内只做内存查找 + 内存更新。
        missed = False
        public_item: dict[str, object] | None = None
        with self._lock:
            matched = find_match(self._items)
            if matched is None:
                # ⚠️ 缓存未命中**也不回读存储**：读库同样要排队等图片任务线程的写锁，
                # 而这里跑在事件循环上。单副本部署下内存快照就是权威
                # （创建/修改/删除密钥都会同步更新 self._items），
                # 回读只对「外部进程直接改库」有意义 —— 交给后台线程兜底。
                self._reload_requested = True
                missed = True
            else:
                index, matched_item = matched
                if touch:
                    # last_used_at 节流：同一密钥 60 秒内只回写一次。
                    throttle_key = self._clean(matched_item.get("id")) or candidate_hash
                    now_ts = time.time()
                    if now_ts - self._last_touch_ts.get(throttle_key, 0.0) >= self._TOUCH_INTERVAL_SECONDS:
                        self._last_touch_ts[throttle_key] = now_ts
                        if len(self._last_touch_ts) > 4096:
                            cutoff = now_ts - self._TOUCH_INTERVAL_SECONDS
                            self._last_touch_ts = {
                                key: ts for key, ts in self._last_touch_ts.items() if ts >= cutoff
                            }
                        # 只更新内存快照 + 记脏，真正的落库交给后台批量 flush。
                        touched_at = datetime.now(timezone.utc).isoformat()
                        next_item = dict(matched_item)
                        next_item["last_used_at"] = touched_at
                        self._items[index] = next_item
                        key_id = self._clean(next_item.get("id"))
                        if key_id:
                            self._pending_touch[key_id] = touched_at
                        matched_item = next_item
                public_item = self._public_item(matched_item)

        if missed:
            # 兜底回读放到后台（每 30 秒一轮），本请求立刻返回 401，不阻塞事件循环。
            self._ensure_touch_flusher()
            return None
        return public_item


auth_service = AuthService(config.get_user_storage_backend())