import asyncio
import hashlib
import json
import time
import unicodedata
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any
from uuid import UUID

import psycopg
from fastapi import Depends, FastAPI, HTTPException, Request, Response, status
import httpx
from fastapi.responses import FileResponse, JSONResponse
from psycopg import Connection
from psycopg.types.json import Json

from .config import settings
from .db import get_connection
from .schemas import (
    AuthResponse,
    LoginRequest,
    MembershipRequest,
    PasswordChangeRequest,
    QuotaAdjustmentRequest,
    RegisterRequest,
    StatusRequest,
    UserPublic,
    VerificationConfirmRequest,
    VerificationSendRequest,
    TtsTaskRequest,
    TtsTaskPublic,
    TtsVoicePublic,
    normalize_email,
)
from .security import (
    check_origin,
    clear_session_cookie,
    code_digest,
    current_user,
    hash_password,
    new_csrf_token,
    new_token,
    password_hasher,
    rate_limiter,
    require_admin,
    require_csrf,
    set_csrf_cookie,
    set_session_cookie,
    token_digest,
    utc_now,
    verify_password,
)

app = FastAPI(title="Kaotings Business API", version="0.2.0", docs_url=None if settings.app_env == "production" else "/docs")


def error(code: str, message: str, http_status: int) -> HTTPException:
    return HTTPException(status_code=http_status, detail={"code": code, "message": message})


def period_bounds(now: datetime | None = None) -> tuple[datetime, datetime]:
    current = now or utc_now()
    start = current.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
    if start.month == 12:
        end = start.replace(year=start.year + 1, month=1)
    else:
        end = start.replace(month=start.month + 1)
    return start, end


def effective_plan(connection: Connection, user: dict) -> str:
    if user["role"] == "admin":
        return user["plan"]
    grant = connection.execute(
        """
        SELECT plan FROM membership_grants
        WHERE user_id = %s AND revoked_at IS NULL AND starts_at <= now()
          AND (expires_at IS NULL OR expires_at > now())
        ORDER BY starts_at DESC LIMIT 1
        """,
        (user["id"],),
    ).fetchone()
    return grant["plan"] if grant else "free"


def user_public(connection: Connection, user: dict) -> UserPublic:
    return UserPublic(
        id=user["id"], email=user["email"], phone=user["phone"], role=user["role"], plan=effective_plan(connection, user),
        status=user["status"], email_verified=user["email_verified"], phone_verified=user["phone_verified"],
        created_at=user["created_at"], last_login_at=user["last_login_at"],
    )


def ensure_quota(connection: Connection, user_id: UUID, plan: str) -> dict:
    start, end = period_bounds()
    policy = connection.execute("SELECT period_limit FROM plan_policies WHERE code = %s", (plan,)).fetchone()
    if not policy:
        raise error("POLICY_MISSING", "权益策略未配置", 500)
    connection.execute(
        """
        INSERT INTO quota_accounts(user_id, period_start, period_end, limit_snapshot)
        VALUES (%s, %s, %s, %s)
        ON CONFLICT (user_id, period_start, period_end) DO NOTHING
        """,
        (user_id, start, end, policy["period_limit"]),
    )
    row = connection.execute(
        "SELECT * FROM quota_accounts WHERE user_id = %s AND period_start = %s AND period_end = %s",
        (user_id, start, end),
    ).fetchone()
    return row


def usage_view(quota: dict) -> dict[str, Any]:
    available = max(0, quota["limit_snapshot"] + quota["adjustment"] - quota["used"] - quota["reserved"])
    return {
        "period_start": quota["period_start"], "period_end": quota["period_end"], "limit": quota["limit_snapshot"],
        "adjustment": quota["adjustment"], "used": quota["used"], "reserved": quota["reserved"], "available": available,
    }


def normalize_tts_text(value: str) -> str:
    return unicodedata.normalize("NFC", value.replace("\r\n", "\n").replace("\r", "\n"))


def task_view(connection: Connection, task: dict) -> dict[str, Any]:
    audio = connection.execute("SELECT expires_at, status FROM audio_files WHERE task_id = %s", (task["id"],)).fetchone()
    return {
        "id": task["id"], "status": task["status"], "text_length": task["text_length"],
        "voice_id": task["provider_voice_id"], "parameters": task["parameters"],
        "error_code": task["error_code"], "audio_available": bool(audio and audio["status"] == "available"),
        "audio_expires_at": audio["expires_at"] if audio else None, "created_at": task["created_at"],
        "started_at": task["started_at"], "finished_at": task["finished_at"],
    }


def claim_next_task() -> dict | None:
    with psycopg.connect(settings.database_url, row_factory=psycopg.rows.dict_row) as connection:
        task = connection.execute("SELECT * FROM tts_tasks WHERE status = 'queued' ORDER BY created_at FOR UPDATE SKIP LOCKED LIMIT 1").fetchone()
        if not task:
            return None
        lease_token = new_token()
        updated = connection.execute("UPDATE tts_tasks SET status = 'running', lease_token = %s, lease_expires_at = now() + interval '5 minutes', started_at = now(), updated_at = now(), attempt_count = attempt_count + 1 WHERE id = %s RETURNING *", (lease_token, task["id"])).fetchone()
        connection.commit()
        return updated


def finish_task_failure(task: dict, code: str) -> None:
    with psycopg.connect(settings.database_url, row_factory=psycopg.rows.dict_row) as connection:
        updated = connection.execute("UPDATE tts_tasks SET status = 'failed', error_code = %s, finished_at = now(), updated_at = now(), lease_token = NULL, lease_expires_at = NULL WHERE id = %s AND status = 'running' AND lease_token = %s", (code, task["id"], task["lease_token"]))
        if updated.rowcount != 1:
            connection.rollback()
            return
        connection.execute("UPDATE quota_accounts SET reserved = GREATEST(0, reserved - %s), version = version + 1 WHERE id = %s", (task["reserved_amount"], task["quota_account_id"]))
        connection.execute("INSERT INTO usage_records(user_id, quota_account_id, task_id, type, amount) VALUES (%s, %s, %s, 'release', %s)", (task["user_id"], task["quota_account_id"], task["id"], task["reserved_amount"]))
        connection.commit()


def finish_task_success(task: dict, audio: bytes, mime_type: str) -> None:
    storage_dir = Path(settings.audio_storage_dir)
    storage_dir.mkdir(parents=True, exist_ok=True)
    extension = "mp3" if mime_type == "audio/mpeg" else "wav"
    storage_key = f"{task['user_id']}/{task['id']}.{extension}"
    target = storage_dir / storage_key
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_bytes(audio)
    checksum = hashlib.sha256(audio).hexdigest()
    with psycopg.connect(settings.database_url, row_factory=psycopg.rows.dict_row) as connection:
        current = connection.execute("SELECT status, lease_token FROM tts_tasks WHERE id = %s FOR UPDATE", (task["id"],)).fetchone()
        if not current or current["status"] != "running" or current["lease_token"] != task["lease_token"]:
            target.unlink(missing_ok=True)
            connection.rollback()
            return
        connection.execute("INSERT INTO audio_files(task_id, owner_id, storage_key, mime_type, size_bytes, checksum) VALUES (%s, %s, %s, %s, %s, %s)", (task["id"], task["user_id"], storage_key, mime_type, len(audio), checksum))
        connection.execute("UPDATE tts_tasks SET status = 'succeeded', finished_at = now(), updated_at = now(), lease_token = NULL, lease_expires_at = NULL WHERE id = %s AND status = 'running' AND lease_token = %s", (task["id"], task["lease_token"]))
        connection.execute("UPDATE quota_accounts SET reserved = GREATEST(0, reserved - %s), used = used + %s, version = version + 1 WHERE id = %s", (task["reserved_amount"], task["reserved_amount"], task["quota_account_id"]))
        connection.execute("INSERT INTO usage_records(user_id, quota_account_id, task_id, type, amount) VALUES (%s, %s, %s, 'consume', %s)", (task["user_id"], task["quota_account_id"], task["id"], task["reserved_amount"]))
        connection.commit()


def recover_expired_tasks() -> None:
    with psycopg.connect(settings.database_url, row_factory=psycopg.rows.dict_row) as connection:
        tasks = connection.execute("SELECT * FROM tts_tasks WHERE status = 'running' AND lease_expires_at IS NOT NULL AND lease_expires_at < now() FOR UPDATE SKIP LOCKED").fetchall()
        for task in tasks:
            connection.execute("UPDATE tts_tasks SET status = 'failed', error_code = 'WORKER_LEASE_EXPIRED', finished_at = now(), updated_at = now(), lease_token = NULL, lease_expires_at = NULL WHERE id = %s AND status = 'running'", (task["id"],))
            connection.execute("UPDATE quota_accounts SET reserved = GREATEST(0, reserved - %s), version = version + 1 WHERE id = %s", (task["reserved_amount"], task["quota_account_id"]))
            connection.execute("INSERT INTO usage_records(user_id, quota_account_id, task_id, type, amount) VALUES (%s, %s, %s, 'release', %s)", (task["user_id"], task["quota_account_id"], task["id"], task["reserved_amount"]))
        connection.commit()


def cleanup_orphan_audio() -> None:
    storage_dir = Path(settings.audio_storage_dir)
    if not storage_dir.exists():
        return
    with psycopg.connect(settings.database_url, row_factory=psycopg.rows.dict_row) as connection:
        known = {row["storage_key"] for row in connection.execute("SELECT storage_key FROM audio_files WHERE status = 'available'").fetchall()}
    now = time.time()
    for path in storage_dir.rglob("*"):
        if path.is_file() and path.relative_to(storage_dir).as_posix() not in known and now - path.stat().st_mtime > 3600:
            path.unlink(missing_ok=True)


async def process_task(task: dict) -> None:
    if not settings.tts_upstream_url:
        await asyncio.to_thread(finish_task_failure, task, "UPSTREAM_NOT_CONFIGURED")
        return
    parameters = task["parameters"]
    response_format = parameters.get("format", "wav")
    payload = {"model": parameters.get("model", "qwen3-tts"), "input": task["text"], "voice": task["provider_voice_id"], "response_format": response_format, "speed": parameters.get("speed", 1.0)}
    headers = {"Content-Type": "application/json"}
    if settings.tts_api_key:
        headers["Authorization"] = f"Bearer {settings.tts_api_key}"
    try:
        async with httpx.AsyncClient(timeout=settings.tts_timeout_seconds) as client:
            response = await client.post(f"{settings.tts_upstream_url}/v1/audio/speech", headers=headers, json=payload)
        if response.status_code == 401:
            raise RuntimeError("UPSTREAM_UNAUTHORIZED")
        if response.status_code >= 400:
            raise RuntimeError(f"UPSTREAM_HTTP_{response.status_code}")
        audio = response.content
        if not audio or len(audio) > 50 * 1024 * 1024:
            raise RuntimeError("UPSTREAM_AUDIO_INVALID")
        mime = "audio/mpeg" if response_format == "mp3" else "audio/wav"
        await asyncio.to_thread(finish_task_success, task, audio, mime)
    except httpx.TimeoutException:
        await asyncio.to_thread(finish_task_failure, task, "UPSTREAM_TIMEOUT")
    except Exception as exc:
        await asyncio.to_thread(finish_task_failure, task, str(exc)[:80])


async def task_worker() -> None:
    while True:
        await asyncio.to_thread(recover_expired_tasks)
        task = await asyncio.to_thread(claim_next_task)
        if task:
            await process_task(task)
        else:
            await asyncio.sleep(1)


worker_task: asyncio.Task | None = None


@app.on_event("startup")
async def start_worker():
    global worker_task
    await asyncio.to_thread(cleanup_orphan_audio)
    worker_task = asyncio.create_task(task_worker())


@app.on_event("shutdown")
async def stop_worker():
    if worker_task:
        worker_task.cancel()


@app.exception_handler(HTTPException)
async def http_error_handler(_: Request, exc: HTTPException):
    detail = exc.detail if isinstance(exc.detail, dict) else {"code": "REQUEST_FAILED", "message": str(exc.detail)}
    return JSONResponse(status_code=exc.status_code, content={"error": detail})


@app.get("/healthz")
def health(connection: Connection = Depends(get_connection)):
    connection.execute("SELECT 1")
    return {"status": "ok", "database": "ok", "service": "api"}


@app.get("/api/v1/auth/csrf")
def csrf(response: Response, request: Request):
    check_origin(request)
    token = request.cookies.get("kaotings_csrf") or new_csrf_token()
    set_csrf_cookie(response, token)
    return {"csrf_token": token}


@app.post("/api/v1/auth/register", response_model=AuthResponse, status_code=201)
def register(payload: RegisterRequest, request: Request, connection: Connection = Depends(get_connection)):
    check_origin(request)
    key = f"register:{request.client.host if request.client else 'unknown'}"
    if not rate_limiter.allow(key, 5, 3600):
        raise error("RATE_LIMITED", "请求过于频繁", 429)
    email = normalize_email(str(payload.email))
    try:
        user = connection.execute(
            """
            INSERT INTO users(email, password_hash) VALUES (%s, %s)
            RETURNING *
            """,
            (email, hash_password(payload.password)),
        ).fetchone()
        ensure_quota(connection, user["id"], "free")
        connection.commit()
    except psycopg.errors.UniqueViolation:
        connection.rollback()
        raise error("REGISTRATION_FAILED", "注册信息不可用", 409)
    return {"user": user_public(connection, user)}


@app.post("/api/v1/auth/login", response_model=AuthResponse)
def login(payload: LoginRequest, request: Request, response: Response, connection: Connection = Depends(get_connection)):
    check_origin(request)
    key = f"login:{request.client.host if request.client else 'unknown'}:{normalize_email(str(payload.email))}"
    if not rate_limiter.allow(key, 10, 300):
        raise error("RATE_LIMITED", "请求过于频繁", 429)
    user = connection.execute("SELECT * FROM users WHERE email = %s", (normalize_email(str(payload.email)),)).fetchone()
    if not user or not verify_password(payload.password, user["password_hash"]):
        raise error("LOGIN_FAILED", "邮箱或密码错误", 401)
    if user["status"] != "active":
        raise error("ACCOUNT_DISABLED", "账户不可用", 403)
    token = new_token()
    connection.execute("INSERT INTO sessions(user_id, token_hash, expires_at) VALUES (%s, %s, %s)", (user["id"], token_digest(token), utc_now() + timedelta(seconds=settings.session_ttl_seconds)))
    user = connection.execute("UPDATE users SET last_login_at = now(), updated_at = now() WHERE id = %s RETURNING *", (user["id"],)).fetchone()
    ensure_quota(connection, user["id"], effective_plan(connection, user))
    connection.commit()
    set_session_cookie(response, token)
    return {"user": user_public(connection, user)}


@app.post("/api/v1/auth/logout", status_code=204, dependencies=[Depends(require_csrf)])
def logout(response: Response, request: Request, connection: Connection = Depends(get_connection)):
    token = request.cookies.get(settings.session_cookie_name)
    if token:
        connection.execute("UPDATE sessions SET revoked_at = now() WHERE token_hash = %s", (token_digest(token),))
        connection.commit()
    clear_session_cookie(response)
    return Response(status_code=204)


@app.get("/api/v1/auth/me", response_model=UserPublic)
def me(user: dict = Depends(current_user), connection: Connection = Depends(get_connection)):
    return user_public(connection, user)


@app.post("/api/v1/auth/password/change", dependencies=[Depends(require_csrf)])
def change_password(payload: PasswordChangeRequest, request: Request, connection: Connection = Depends(get_connection), user: dict = Depends(current_user)):
    if not verify_password(payload.current_password, user["password_hash"]):
        raise error("PASSWORD_INVALID", "当前密码错误", 400)
    connection.execute("UPDATE users SET password_hash = %s, updated_at = now() WHERE id = %s", (hash_password(payload.new_password), user["id"]))
    connection.execute("UPDATE sessions SET revoked_at = now() WHERE user_id = %s AND id <> %s", (user["id"], user["session_id"]))
    connection.commit()
    return {"status": "ok"}


@app.get("/api/v1/account/usage")
def account_usage(user: dict = Depends(current_user), connection: Connection = Depends(get_connection)):
    quota = ensure_quota(connection, user["id"], effective_plan(connection, user))
    connection.commit()
    return {"plan": effective_plan(connection, user), **usage_view(quota)}


@app.post("/api/v1/auth/verification/send")
def verification_send(payload: VerificationSendRequest, request: Request, user: dict = Depends(current_user)):
    check_origin(request)
    enabled = settings.email_verification_enabled if payload.channel == "email" else settings.phone_verification_enabled
    if not enabled:
        raise error("VERIFICATION_NOT_ENABLED", "验证渠道尚未启用", 503)
    raise error("VERIFICATION_PROVIDER_UNAVAILABLE", "验证渠道尚未配置", 503)


@app.post("/api/v1/auth/verification/confirm", dependencies=[Depends(require_csrf)])
def verification_confirm(payload: VerificationConfirmRequest, request: Request, connection: Connection = Depends(get_connection), user: dict = Depends(current_user)):
    check_origin(request)
    challenge = connection.execute("SELECT * FROM verification_challenges WHERE id = %s AND user_id = %s", (payload.challenge_id, user["id"])).fetchone()
    if not challenge or challenge["consumed_at"] or challenge["expires_at"] <= utc_now() or challenge["attempt_count"] >= 5:
        raise error("VERIFICATION_INVALID", "验证挑战无效或已过期", 400)
    if not __import__("hmac").compare_digest(challenge["code_digest"], code_digest(payload.code)):
        connection.execute("UPDATE verification_challenges SET attempt_count = attempt_count + 1 WHERE id = %s", (payload.challenge_id,))
        connection.commit()
        raise error("VERIFICATION_INVALID", "验证码错误", 400)
    connection.execute("UPDATE verification_challenges SET consumed_at = now() WHERE id = %s", (payload.challenge_id,))
    if challenge["channel"] == "email":
        connection.execute("UPDATE users SET email_verified = true, updated_at = now() WHERE id = %s", (user["id"],))
    else:
        connection.execute("UPDATE users SET phone_verified = true, updated_at = now() WHERE id = %s", (user["id"],))
    connection.commit()
    return {"status": "ok"}


@app.get("/api/v1/tts/voices", response_model=list[TtsVoicePublic])
def tts_voices(connection: Connection = Depends(get_connection)):
    return connection.execute("SELECT id, provider_voice_id, name, language, description, supported_parameters FROM tts_voices WHERE enabled = true ORDER BY name").fetchall()


@app.post("/api/v1/tts/tasks", status_code=202)
def create_tts_task(payload: TtsTaskRequest, request: Request, connection: Connection = Depends(get_connection), user: dict = Depends(current_user), _: None = Depends(require_csrf)):
    idempotency_key = request.headers.get("idempotency-key", "").strip()
    if len(idempotency_key) < 8 or len(idempotency_key) > 120:
        raise error("IDEMPOTENCY_REQUIRED", "需要有效的 Idempotency-Key", 400)
    plan = effective_plan(connection, user)
    policy = connection.execute("SELECT * FROM plan_policies WHERE code = %s", (plan,)).fetchone()
    text = normalize_tts_text(payload.text)
    if not text.strip():
        raise error("EMPTY_TEXT", "文本不能只有空白", 422)
    text_length = len(text)
    if not policy or text_length > policy["max_text_length"]:
        raise error("TEXT_TOO_LONG", "文本超过当前计划限制", 422)
    parameters = dict(payload.parameters)
    response_format = str(parameters.get("format", "wav")).lower()
    speed = float(parameters.get("speed", 1.0))
    if response_format != "wav" or speed != 1:
        raise error("UNVERIFIED_PARAMETERS", "V1 当前只支持 WAV 和 1.0x 语速", 422)
    voice = connection.execute("SELECT * FROM tts_voices WHERE provider_voice_id = %s AND enabled = true", (payload.voice_id,)).fetchone()
    if not voice or plan not in (voice["allowed_plans"] or ["free", "vip"]):
        raise error("VOICE_NOT_ALLOWED", "音色不可用", 422)
    request_hash = hashlib.sha256(json.dumps({"text": text, "voice": payload.voice_id, "parameters": parameters}, ensure_ascii=False, sort_keys=True).encode()).hexdigest()
    connection.execute("SELECT pg_advisory_xact_lock(hashtext(%s))", (str(user["id"]),))
    existing = connection.execute("SELECT * FROM tts_tasks WHERE user_id = %s AND idempotency_key = %s", (user["id"], idempotency_key)).fetchone()
    if existing:
        if existing["request_hash"] != request_hash:
            raise error("IDEMPOTENCY_CONFLICT", "幂等键已用于其他请求", 409)
        return task_view(connection, existing)
    active_count = connection.execute("SELECT count(*) FROM tts_tasks WHERE user_id = %s AND status IN ('queued', 'running')", (user["id"],)).fetchone()["count"]
    if active_count >= policy["max_concurrency"]:
        raise error("CONCURRENCY_LIMIT", "并发任务数已达上限", 409)
    quota = ensure_quota(connection, user["id"], plan)
    quota = connection.execute("SELECT * FROM quota_accounts WHERE id = %s FOR UPDATE", (quota["id"],)).fetchone()
    available = quota["limit_snapshot"] + quota["adjustment"] - quota["used"] - quota["reserved"]
    if available < text_length:
        raise error("QUOTA_EXCEEDED", "额度不足", 409)
    task = connection.execute(
        """
        INSERT INTO tts_tasks(user_id, text, text_length, voice_id, provider_voice_id, parameters, idempotency_key, request_hash, policy_version, quota_account_id, reserved_amount)
        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) RETURNING *
        """,
        (user["id"], text, text_length, voice["id"], payload.voice_id, Json(parameters), idempotency_key, request_hash, policy["version"], quota["id"], text_length),
    ).fetchone()
    connection.execute("UPDATE quota_accounts SET reserved = reserved + %s, version = version + 1 WHERE id = %s", (text_length, quota["id"]))
    connection.execute("INSERT INTO usage_records(user_id, quota_account_id, task_id, type, amount, idempotency_key) VALUES (%s, %s, %s, 'reserve', %s, %s)", (user["id"], quota["id"], task["id"], text_length, f"reserve:{task['id']}"))
    connection.commit()
    return task_view(connection, task)


@app.get("/api/v1/tts/tasks", response_model=list[TtsTaskPublic])
def list_tts_tasks(user: dict = Depends(current_user), connection: Connection = Depends(get_connection)):
    tasks = connection.execute("SELECT * FROM tts_tasks WHERE user_id = %s ORDER BY created_at DESC LIMIT 100", (user["id"],)).fetchall()
    return [task_view(connection, task) for task in tasks]


def owned_task(task_id: UUID, user: dict, connection: Connection) -> dict:
    task = connection.execute("SELECT * FROM tts_tasks WHERE id = %s", (task_id,)).fetchone()
    if not task or (task["user_id"] != user["id"] and user["role"] != "admin"):
        raise error("NOT_FOUND", "任务不存在", 404)
    return task


@app.get("/api/v1/tts/tasks/{task_id}", response_model=TtsTaskPublic)
def get_tts_task(task_id: UUID, user: dict = Depends(current_user), connection: Connection = Depends(get_connection)):
    return task_view(connection, owned_task(task_id, user, connection))


def audio_response(task_id: UUID, user: dict, connection: Connection, download: bool):
    task = owned_task(task_id, user, connection)
    audio = connection.execute("SELECT * FROM audio_files WHERE task_id = %s AND status = 'available'", (task_id,)).fetchone()
    if task["status"] != "succeeded" or not audio:
        raise error("AUDIO_NOT_AVAILABLE", "音频尚不可用", 404)
    path = Path(settings.audio_storage_dir) / audio["storage_key"]
    if not path.is_file():
        raise error("AUDIO_NOT_AVAILABLE", "音频文件不可用", 404)
    filename = f"kaotings-{task_id}.{'mp3' if audio['mime_type'] == 'audio/mpeg' else 'wav'}" if download else None
    return FileResponse(path, media_type=audio["mime_type"], filename=filename)


@app.get("/api/v1/tts/tasks/{task_id}/audio")
def play_tts_audio(task_id: UUID, user: dict = Depends(current_user), connection: Connection = Depends(get_connection)):
    return audio_response(task_id, user, connection, False)


@app.get("/api/v1/tts/tasks/{task_id}/download")
def download_tts_audio(task_id: UUID, user: dict = Depends(current_user), connection: Connection = Depends(get_connection)):
    return audio_response(task_id, user, connection, True)


@app.get("/api/v1/admin/users")
def admin_users(user: dict = Depends(require_admin), connection: Connection = Depends(get_connection)):
    rows = connection.execute("SELECT id, email, phone, role, plan, status, email_verified, phone_verified, created_at, last_login_at FROM users ORDER BY created_at DESC LIMIT 100").fetchall()
    return {"items": rows}


@app.patch("/api/v1/admin/users/{user_id}/status", dependencies=[Depends(require_csrf)])
def admin_status(user_id: UUID, payload: StatusRequest, connection: Connection = Depends(get_connection), actor: dict = Depends(require_admin)):
    target = connection.execute("SELECT * FROM users WHERE id = %s", (user_id,)).fetchone()
    if not target:
        raise error("NOT_FOUND", "用户不存在", 404)
    if target["role"] == "admin" and payload.status == "disabled":
        count = connection.execute("SELECT count(*) FROM users WHERE role = 'admin' AND status = 'active'").fetchone()["count"]
        if count <= 1:
            raise error("LAST_ADMIN_PROTECTED", "不能禁用最后一个可用管理员", 409)
    updated = connection.execute("UPDATE users SET status = %s, updated_at = now() WHERE id = %s RETURNING id, status", (payload.status, user_id)).fetchone()
    connection.execute("INSERT INTO admin_audit_logs(actor_id, action, target_type, target_id, before_value, after_value, reason) VALUES (%s, 'user_status', 'user', %s, %s, %s, %s)", (actor["id"], user_id, Json({"status": target["status"]}), Json({"status": updated["status"]}), payload.reason))
    connection.commit()
    return updated


@app.put("/api/v1/admin/users/{user_id}/membership", dependencies=[Depends(require_csrf)])
def admin_membership(user_id: UUID, payload: MembershipRequest, connection: Connection = Depends(get_connection), actor: dict = Depends(require_admin)):
    target = connection.execute("SELECT * FROM users WHERE id = %s", (user_id,)).fetchone()
    if not target:
        raise error("NOT_FOUND", "用户不存在", 404)
    starts_at = payload.starts_at or utc_now()
    connection.execute("UPDATE membership_grants SET revoked_at = now() WHERE user_id = %s AND revoked_at IS NULL", (user_id,))
    grant = connection.execute("INSERT INTO membership_grants(user_id, plan, starts_at, expires_at, created_by, reason) VALUES (%s, %s, %s, %s, %s, %s) RETURNING id, plan, starts_at, expires_at", (user_id, payload.plan, starts_at, payload.expires_at, actor["id"], payload.reason)).fetchone()
    connection.execute("UPDATE users SET plan = %s, updated_at = now() WHERE id = %s", (payload.plan, user_id))
    connection.execute("INSERT INTO admin_audit_logs(actor_id, action, target_type, target_id, before_value, after_value, reason) VALUES (%s, 'membership', 'user', %s, %s, %s, %s)", (actor["id"], user_id, Json({"plan": target["plan"]}), Json({"plan": payload.plan, "expires_at": payload.expires_at.isoformat() if payload.expires_at else None}), payload.reason))
    connection.commit()
    return grant


@app.post("/api/v1/admin/users/{user_id}/quota-adjustments", dependencies=[Depends(require_csrf)])
def admin_quota(user_id: UUID, payload: QuotaAdjustmentRequest, connection: Connection = Depends(get_connection), actor: dict = Depends(require_admin)):
    target = connection.execute("SELECT * FROM users WHERE id = %s", (user_id,)).fetchone()
    if not target:
        raise error("NOT_FOUND", "用户不存在", 404)
    quota = ensure_quota(connection, user_id, effective_plan(connection, target))
    existing = connection.execute("SELECT id FROM usage_records WHERE user_id = %s AND idempotency_key = %s", (user_id, payload.idempotency_key)).fetchone()
    if existing:
        raise error("IDEMPOTENCY_CONFLICT", "该调整已提交", 409)
    connection.execute("UPDATE quota_accounts SET adjustment = adjustment + %s, version = version + 1 WHERE id = %s", (payload.amount, quota["id"]))
    connection.execute("INSERT INTO usage_records(user_id, quota_account_id, type, amount, idempotency_key) VALUES (%s, %s, 'adjust', %s, %s)", (user_id, quota["id"], payload.amount, payload.idempotency_key))
    connection.execute("INSERT INTO admin_audit_logs(actor_id, action, target_type, target_id, after_value, reason) VALUES (%s, 'quota_adjustment', 'user', %s, %s, %s)", (actor["id"], user_id, Json({"amount": payload.amount}), payload.reason))
    connection.commit()
    return {"status": "ok", "amount": payload.amount}