import asyncio
import hashlib
import json
import secrets
import smtplib
import ssl
import time
import unicodedata
from datetime import datetime, timedelta, timezone
from email.message import EmailMessage
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,
    MembershipRevokeRequest,
    PasswordChangeRequest,
    QuotaAdjustmentRequest,
    RegisterRequest,
    RegisterResponse,
    StatusRequest,
    UserPublic,
    VerificationConfirmRequest,
    VerificationSendRequest,
    VerificationResendRequest,
    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 send_verification_email(target: str, code: str) -> None:
    if not all((settings.smtp_host, settings.smtp_user, settings.smtp_password, settings.smtp_from)):
        raise RuntimeError("SMTP_NOT_CONFIGURED")
    message = EmailMessage()
    message["From"] = settings.smtp_from
    message["To"] = target
    message["Subject"] = "考町科技邮箱验证码"
    message.set_content(f"您好，您的考町科技邮箱验证码是：{code}\n\n验证码 10 分钟内有效。如非本人操作，请忽略此邮件。")
    with smtplib.SMTP_SSL(settings.smtp_host, settings.smtp_port, context=ssl.create_default_context(), timeout=20) as smtp:
        smtp.login(settings.smtp_user, settings.smtp_password)
        smtp.send_message(message)


def create_email_challenge(connection: Connection, user_id: UUID, email: str) -> tuple[UUID, str]:
    code = f"{secrets.randbelow(1_000_000):06d}"
    row = connection.execute(
        "INSERT INTO verification_challenges(user_id, channel, purpose, target, code_digest, expires_at) VALUES (%s, 'email', 'registration', %s, %s, now() + interval '10 minutes') RETURNING id",
        (user_id, email, code_digest(code)),
    ).fetchone()
    return row["id"], code


def send_sms_verification(phone: str) -> str:
    if not all((settings.aliyun_access_key_id, settings.aliyun_access_key_secret, settings.aliyun_sign_name, settings.aliyun_template_code)):
        raise RuntimeError("ALIYUN_PNVS_NOT_CONFIGURED")
    from alibabacloud_dypnsapi20170525.client import Client
    from alibabacloud_dypnsapi20170525 import models as dypns_models
    from alibabacloud_tea_openapi import models as openapi_models
    from alibabacloud_tea_util import models as util_models

    client_config = openapi_models.Config(
        access_key_id=settings.aliyun_access_key_id,
        access_key_secret=settings.aliyun_access_key_secret,
        endpoint="dypnsapi.aliyuncs.com",
    )
    client = Client(client_config)
    out_id = secrets.token_urlsafe(24)
    request = dypns_models.SendSmsVerifyCodeRequest(
        phone_number=phone,
        sign_name=settings.aliyun_sign_name,
        template_code=settings.aliyun_template_code,
        scheme_name=settings.aliyun_scheme_name or None,
        country_code="86",
        template_param='{"code":"##code##","min":"5"}',
        out_id=out_id,
        return_verify_code=False,
    )
    response = client.send_sms_verify_code_with_options(request, util_models.RuntimeOptions())
    body = getattr(response, "body", response)
    if getattr(body, "code", "") != "OK" or not getattr(body, "success", False):
        raise RuntimeError(f"ALIYUN_PNVS_SEND_FAILED:{getattr(body, 'code', 'UNKNOWN')}")
    return out_id


def check_sms_verification(phone: str, code: str, out_id: str) -> bool:
    from alibabacloud_dypnsapi20170525.client import Client
    from alibabacloud_dypnsapi20170525 import models as dypns_models
    from alibabacloud_tea_openapi import models as openapi_models
    from alibabacloud_tea_util import models as util_models

    client = Client(openapi_models.Config(
        access_key_id=settings.aliyun_access_key_id,
        access_key_secret=settings.aliyun_access_key_secret,
        endpoint="dypnsapi.aliyuncs.com",
    ))
    request = dypns_models.CheckSmsVerifyCodeRequest(
        phone_number=phone,
        verify_code=code,
        country_code="86",
        scheme_name=settings.aliyun_scheme_name or None,
        out_id=out_id,
        case_auth_policy=1,
    )
    response = client.check_sms_verify_code_with_options(request, util_models.RuntimeOptions())
    body = getattr(response, "body", response)
    model = getattr(body, "model", None)
    return getattr(model, "verify_result", "") == "PASS"


def create_sms_challenge(connection: Connection, user_id: UUID, phone: str) -> UUID:
    out_id = send_sms_verification(phone)
    row = connection.execute(
        """
        INSERT INTO verification_challenges(user_id, channel, purpose, target, code_digest, provider, provider_request_id, expires_at)
        VALUES (%s, 'phone', 'registration', %s, %s, 'aliyun_pnvs', %s, now() + interval '10 minutes') RETURNING id
        """,
        (user_id, phone, code_digest(out_id), out_id),
    ).fetchone()
    return row["id"]


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("SELECT id FROM quota_accounts WHERE id = %s FOR UPDATE", (task["quota_account_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 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("SELECT id FROM quota_accounts WHERE id = %s FOR UPDATE", (task["quota_account_id"],))
        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("SELECT id FROM quota_accounts WHERE id = %s FOR UPDATE", (task["quota_account_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}")
        content_type = response.headers.get("content-type", "").split(";", 1)[0].lower()
        if content_type not in {"audio/wav", "audio/x-wav"}:
            raise RuntimeError("UPSTREAM_NOT_AUDIO")
        declared_size = response.headers.get("content-length")
        if declared_size and int(declared_size) > 50 * 1024 * 1024:
            raise RuntimeError("UPSTREAM_RESPONSE_TOO_LARGE")
        audio = response.content
        if not audio or len(audio) > 50 * 1024 * 1024:
            raise RuntimeError("UPSTREAM_AUDIO_INVALID")
        if response_format == "wav" and (len(audio) < 12 or audio[:4] != b"RIFF" or audio[8:12] != b"WAVE"):
            raise RuntimeError("UPSTREAM_AUDIO_CORRUPT")
        await asyncio.to_thread(finish_task_success, task, audio, "audio/wav")
    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=RegisterResponse, 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)) if payload.email else None
    phone = payload.phone
    try:
        user = connection.execute(
            """
            INSERT INTO users(email, phone, password_hash) VALUES (%s, %s, %s)
            RETURNING *
            """,
            (email, phone, hash_password(payload.password)),
        ).fetchone()
        ensure_quota(connection, user["id"], "free")
        if payload.verification_channel == "email":
            challenge_id, code = create_email_challenge(connection, user["id"], email)
        else:
            challenge_id = create_sms_challenge(connection, user["id"], phone)
            code = None
    except psycopg.errors.UniqueViolation:
        connection.rollback()
        raise error("REGISTRATION_FAILED", "注册信息不可用", 409)
    except Exception:
        connection.rollback()
        raise error("VERIFICATION_SEND_FAILED", "验证码发送失败，请稍后重试", 503)
    try:
        if payload.verification_channel == "email":
            send_verification_email(email, code)
        connection.commit()
    except Exception:
        connection.rollback()
        raise error("VERIFICATION_SEND_FAILED", "验证码发送失败，请稍后重试", 503)
    return {"user": user_public(connection, user), "verification_required": True, "challenge_id": challenge_id}


@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)
    identifier = payload.identifier.strip().lower()
    key = f"login:{request.client.host if request.client else 'unknown'}:{identifier}"
    if not rate_limiter.allow(key, 10, 300):
        raise error("RATE_LIMITED", "请求过于频繁", 429)
    user = connection.execute("SELECT * FROM users WHERE lower(email) = %s OR phone = %s", (identifier, identifier)).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)
    if not (user["email_verified"] or user["phone_verified"]):
        raise error("CONTACT_NOT_VERIFIED", "请先完成邮箱或手机验证", 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, connection: Connection = Depends(get_connection), 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)
    if payload.channel != "email" or payload.purpose != "registration":
        raise error("VERIFICATION_PROVIDER_UNAVAILABLE", "该验证流程尚未配置", 503)
    code = f"{secrets.randbelow(1_000_000):06d}"
    challenge = user
    row = connection.execute("SELECT id FROM verification_challenges WHERE user_id = %s AND channel = 'email' AND purpose = 'registration' AND consumed_at IS NULL ORDER BY created_at DESC LIMIT 1", (challenge["id"],)).fetchone()
    if not row:
        raise error("VERIFICATION_INVALID", "验证挑战无效", 400)
    connection.execute("UPDATE verification_challenges SET code_digest = %s, expires_at = now() + interval '10 minutes', attempt_count = 0 WHERE id = %s", (code_digest(code), row["id"]))
    connection.commit()
    try:
        send_verification_email(challenge["email"], code)
    except Exception:
        raise error("VERIFICATION_EMAIL_FAILED", "验证码邮件发送失败，请稍后重试", 503)
    return {"challenge_id": row["id"], "status": "sent"}


@app.post("/api/v1/auth/verification/resend")
def verification_resend(payload: VerificationResendRequest, request: Request, connection: Connection = Depends(get_connection)):
    check_origin(request)
    challenge = connection.execute("SELECT * FROM verification_challenges WHERE id = %s AND purpose = 'registration' AND consumed_at IS NULL", (payload.challenge_id,)).fetchone()
    if not challenge:
        raise error("VERIFICATION_INVALID", "验证挑战无效", 400)
    if not rate_limiter.allow(f"verification:{challenge['target']}", 3, 900):
        raise error("RATE_LIMITED", "验证码发送过于频繁", 429)
    try:
        if challenge["channel"] == "email":
            code = f"{secrets.randbelow(1_000_000):06d}"
            connection.execute("UPDATE verification_challenges SET code_digest = %s, expires_at = now() + interval '10 minutes', attempt_count = 0 WHERE id = %s", (code_digest(code), payload.challenge_id))
            connection.commit()
            send_verification_email(challenge["target"], code)
        else:
            out_id = send_sms_verification(challenge["target"])
            connection.execute("UPDATE verification_challenges SET code_digest = %s, provider_request_id = %s, expires_at = now() + interval '10 minutes', attempt_count = 0 WHERE id = %s", (code_digest(out_id), out_id, payload.challenge_id))
            connection.commit()
    except Exception:
        connection.rollback()
        raise error("VERIFICATION_SEND_FAILED", "验证码发送失败，请稍后重试", 503)
    return {"challenge_id": payload.challenge_id, "status": "sent"}


@app.post("/api/v1/auth/verification/confirm-registration", dependencies=[Depends(require_csrf)])
def verification_confirm_registration(payload: VerificationConfirmRequest, request: Request, connection: Connection = Depends(get_connection)):
    check_origin(request)
    challenge = connection.execute("SELECT * FROM verification_challenges WHERE id = %s AND purpose = 'registration'", (payload.challenge_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)
    try:
        valid = check_sms_verification(challenge["target"], payload.code, challenge["provider_request_id"]) if challenge["channel"] == "phone" else __import__("hmac").compare_digest(challenge["code_digest"], code_digest(payload.code))
    except Exception:
        raise error("VERIFICATION_PROVIDER_FAILED", "验证码服务暂时不可用，请稍后重试", 503)
    if not valid:
        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,))
    column = "email_verified" if challenge["channel"] == "email" else "phone_verified"
    connection.execute(f"UPDATE users SET {column} = true, updated_at = now() WHERE id = %s", (challenge["user_id"],))
    connection.commit()
    return {"status": "ok"}


@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)


USER_COLUMNS = "id, email, phone, role, plan, status, email_verified, phone_verified, created_at, last_login_at, updated_at"


def quota_view_for(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 active_grant(connection: Connection, user_id: UUID) -> dict | None:
    return connection.execute(
        """
        SELECT id, plan, starts_at, expires_at, reason, created_at
        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()


@app.get("/api/v1/admin/users")
def admin_users(email: str | None = None, phone: str | None = None, status: str | None = None, page: int = 1, limit: int = 20, user: dict = Depends(require_admin), connection: Connection = Depends(get_connection)):
    page = max(1, page)
    limit = min(100, max(1, limit))
    clauses = []
    params: list[Any] = []
    if email:
        clauses.append("email ILIKE %s")
        params.append(f"%{email}%")
    if phone:
        clauses.append("phone ILIKE %s")
        params.append(f"%{phone}%")
    if status in ("active", "disabled"):
        clauses.append("status = %s")
        params.append(status)
    where = " WHERE " + " AND ".join(clauses) if clauses else ""
    total = connection.execute(f"SELECT count(*) AS c FROM users{where}", params).fetchone()["c"]
    rows = connection.execute(
        f"SELECT {USER_COLUMNS} FROM users{where} ORDER BY created_at DESC LIMIT %s OFFSET %s",
        [*params, limit, (page - 1) * limit],
    ).fetchall()
    items = []
    for row in rows:
        plan = effective_plan(connection, row)
        item = dict(row)
        item["effective_plan"] = plan
        items.append(item)
    return {"items": items, "total": total, "page": page, "limit": limit, "pages": (total + limit - 1) // limit}


@app.get("/api/v1/admin/users/{user_id}")
def admin_user_detail(user_id: UUID, user: dict = Depends(require_admin), connection: Connection = Depends(get_connection)):
    target = connection.execute(f"SELECT {USER_COLUMNS} FROM users WHERE id = %s", (user_id,)).fetchone()
    if not target:
        raise error("NOT_FOUND", "用户不存在", 404)
    plan = effective_plan(connection, target)
    quota = ensure_quota(connection, user_id, plan)
    grant = active_grant(connection, user_id)
    grant_history = connection.execute("SELECT id, plan, starts_at, expires_at, revoked_at, reason, created_at FROM membership_grants WHERE user_id = %s ORDER BY created_at DESC", (user_id,)).fetchall()
    adjustment_records = connection.execute(
        """
        SELECT u.id, u.amount, u.idempotency_key, u.created_at, a.reason, a.actor_id
        FROM usage_records u
        LEFT JOIN admin_audit_logs a ON a.action = 'quota_adjustment' AND a.target_id = %s
        WHERE u.user_id = %s AND u.type = 'adjust'
        ORDER BY u.created_at DESC
        """,
        (user_id, user_id),
    ).fetchall()
    connection.commit()
    return {
        "user": dict(target), "effective_plan": plan, "effective_grant": grant,
        "quota": quota_view_for(quota), "membership_history": grant_history, "adjustment_records": adjustment_records,
    }


@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 and target["status"] == "active":
            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()
    if payload.status == "disabled":
        revoked = connection.execute("UPDATE sessions SET revoked_at = now() WHERE user_id = %s AND revoked_at IS NULL", (user_id,))
        revoked_count = revoked.rowcount
    else:
        revoked_count = 0
    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 {"id": updated["id"], "status": updated["status"], "sessions_revoked": revoked_count}


@app.get("/api/v1/admin/users/{user_id}/membership")
def admin_membership_view(user_id: UUID, user: dict = Depends(require_admin), connection: Connection = Depends(get_connection)):
    target = connection.execute("SELECT id, role, plan FROM users WHERE id = %s", (user_id,)).fetchone()
    if not target:
        raise error("NOT_FOUND", "用户不存在", 404)
    connection.commit()
    return {"user_id": user_id, "effective_plan": effective_plan(connection, target), "effective_grant": active_grant(connection, user_id)}


@app.post("/api/v1/admin/users/{user_id}/membership/revoke", dependencies=[Depends(require_csrf)])
def admin_membership_revoke(user_id: UUID, payload: MembershipRevokeRequest, 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)
    grant = connection.execute("SELECT id, plan, revoked_at 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()
    if not grant:
        raise error("NO_ACTIVE_GRANT", "没有可撤销的有效会员", 409)
    connection.execute("UPDATE membership_grants SET revoked_at = now() WHERE id = %s", (grant["id"],))
    connection.execute("UPDATE users SET plan = 'free', updated_at = now() WHERE id = %s", (user_id,))
    connection.execute("INSERT INTO admin_audit_logs(actor_id, action, target_type, target_id, before_value, after_value, reason) VALUES (%s, 'membership_revoke', 'user', %s, %s, %s, %s)", (actor["id"], user_id, Json({"plan": grant["plan"]}), Json({"plan": "free"}), payload.reason))
    connection.commit()
    return {"status": "ok", "revoked_plan": grant["plan"]}


@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()
    if payload.expires_at and payload.expires_at <= starts_at:
        raise error("INVALID_PERIOD", "过期时间必须在开始时间之后", 422)
    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()
    stored_plan = "free" if payload.plan == "free" else payload.plan
    connection.execute("UPDATE users SET plan = %s, updated_at = now() WHERE id = %s", (stored_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.get("/api/v1/admin/users/{user_id}/quota")
def admin_quota_view(user_id: UUID, user: dict = Depends(require_admin), connection: Connection = Depends(get_connection)):
    target = connection.execute("SELECT id, role, plan 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))
    adjustment_records = connection.execute(
        """
        SELECT u.id, u.amount, u.idempotency_key, u.created_at, a.reason, a.actor_id
        FROM usage_records u
        LEFT JOIN admin_audit_logs a ON a.action = 'quota_adjustment' AND a.target_id = %s
        WHERE u.user_id = %s AND u.type = 'adjust'
        ORDER BY u.created_at DESC
        """,
        (user_id, user_id),
    ).fetchall()
    connection.commit()
    return {"quota": quota_view_for(quota), "adjustment_records": adjustment_records}


@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))
    # Serialize with task freeze (create_tts_task) and settlement on the same
    # user, then re-read the quota row under a row lock so the projected balance
    # check is transactionally consistent. The check must cover BOTH used and
    # reserved (frozen) amounts, not only used.
    connection.execute("SELECT pg_advisory_xact_lock(hashtext(%s))", (str(user_id),))
    quota = connection.execute("SELECT * FROM quota_accounts WHERE id = %s FOR UPDATE", (quota["id"],)).fetchone()
    existing = connection.execute("SELECT id FROM usage_records WHERE user_id = %s AND idempotency_key = %s AND type = 'adjust'", (user_id, payload.idempotency_key)).fetchone()
    if existing:
        raise error("IDEMPOTENCY_CONFLICT", "该调整已提交", 409)
    projected = quota["limit_snapshot"] + quota["adjustment"] + payload.amount - quota["used"] - quota["reserved"]
    if projected < 0:
        raise error("QUOTA_BALANCE_INVALID", "调整后额度不能小于已用或冻结额度", 422)
    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}


@app.get("/api/v1/admin/tts/tasks")
def admin_tts_tasks(user_id: UUID | None = None, status: str | None = None, created_after: datetime | None = None, created_before: datetime | None = None, page: int = 1, limit: int = 20, user: dict = Depends(require_admin), connection: Connection = Depends(get_connection)):
    page = max(1, page)
    limit = min(100, max(1, limit))
    clauses = []
    params: list[Any] = []
    if user_id:
        clauses.append("t.user_id = %s")
        params.append(user_id)
    if status in ("queued", "running", "succeeded", "failed"):
        clauses.append("t.status = %s")
        params.append(status)
    if created_after:
        clauses.append("t.created_at >= %s")
        params.append(created_after)
    if created_before:
        clauses.append("t.created_at <= %s")
        params.append(created_before)
    where = " WHERE " + " AND ".join(clauses) if clauses else ""
    total = connection.execute(f"SELECT count(*) AS c FROM tts_tasks t{where}", params).fetchone()["c"]
    rows = connection.execute(
        f"""
        SELECT t.id, t.user_id, u.email AS user_email, t.status, t.text_length, t.provider_voice_id,
               t.error_code, t.attempt_count, t.reserved_amount, t.created_at, t.started_at, t.finished_at,
               (EXTRACT(EPOCH FROM (t.finished_at - t.started_at)) * 1000)::bigint AS duration_ms,
               CASE WHEN t.status = 'succeeded' THEN 'consumed' WHEN t.status = 'failed' THEN 'released'
                    ELSE 'pending' END AS settlement,
               CASE WHEN t.status IN ('queued', 'running') THEN t.reserved_amount ELSE 0 END AS reserved,
               CASE WHEN t.status = 'succeeded' THEN t.reserved_amount ELSE 0 END AS used
        FROM tts_tasks t JOIN users u ON u.id = t.user_id{where}
        ORDER BY t.created_at DESC LIMIT %s OFFSET %s
        """,
        [*params, limit, (page - 1) * limit],
    ).fetchall()
    return {"items": rows, "total": total, "page": page, "limit": limit, "pages": (total + limit - 1) // limit}


@app.get("/api/v1/admin/usage/summary")
def admin_usage_summary(user: dict = Depends(require_admin), connection: Connection = Depends(get_connection)):
    total_used = connection.execute("SELECT COALESCE(sum(used), 0)::bigint AS total FROM quota_accounts").fetchone()["total"]
    total_reserved = connection.execute("SELECT COALESCE(sum(reserved), 0)::bigint AS total FROM quota_accounts").fetchone()["total"]
    total_adjust = connection.execute("SELECT COALESCE(sum(adjustment), 0)::bigint AS total FROM quota_accounts").fetchone()["total"]
    record_total = connection.execute("SELECT COALESCE(sum(CASE WHEN type IN ('consume', 'adjust') THEN amount WHEN type = 'release' THEN -amount ELSE 0 END), 0)::bigint AS total FROM usage_records").fetchone()["total"]
    task_counts = connection.execute("SELECT status, count(*) AS c FROM tts_tasks GROUP BY status").fetchall()
    by_status = {row["status"]: row["c"] for row in task_counts}
    failed = connection.execute("SELECT count(*) AS c FROM tts_tasks WHERE status = 'failed'").fetchone()["c"]
    return {
        "quota": {"used": total_used, "reserved": total_reserved, "adjustment": total_adjust},
        "ledger_total": record_total,
        "tasks": {"total": sum(by_status.values()), "by_status": by_status, "failed": failed},
    }


@app.get("/api/v1/admin/audit-logs")
def admin_audit_logs(target_type: str | None = None, actor_id: UUID | None = None, page: int = 1, limit: int = 30, user: dict = Depends(require_admin), connection: Connection = Depends(get_connection)):
    page = max(1, page)
    limit = min(100, max(1, limit))
    clauses = []
    params: list[Any] = []
    if target_type in ("user", "quota_account", "product", "tts_task"):
        clauses.append("a.target_type = %s")
        params.append(target_type)
    if actor_id:
        clauses.append("a.actor_id = %s")
        params.append(actor_id)
    where = " WHERE " + " AND ".join(clauses) if clauses else ""
    total = connection.execute(f"SELECT count(*) AS c FROM admin_audit_logs a{where}", params).fetchone()["c"]
    rows = connection.execute(
        f"""
        SELECT a.id, a.actor_id, u.email AS actor_email, a.action, a.target_type, a.target_id,
               a.before_value, a.after_value, a.reason, a.created_at
        FROM admin_audit_logs a LEFT JOIN users u ON u.id = a.actor_id{where}
        ORDER BY a.created_at DESC LIMIT %s OFFSET %s
        """,
        [*params, limit, (page - 1) * limit],
    ).fetchall()
    return {"items": rows, "total": total, "page": page, "limit": limit, "pages": (total + limit - 1) // limit}