"""Authentication endpoints."""

import asyncio
import json
import logging
import os
import time
from ipaddress import ip_address, ip_network
from typing import Literal

import httpx
from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response, status
from fastapi.security import OAuth2PasswordRequestForm
from pydantic import BaseModel, ConfigDict, EmailStr, Field, TypeAdapter, ValidationError, field_validator, model_validator

from app.gateway.admin_query_window import recent_admin_query_cutoff
from app.gateway.auth import (
    UserResponse,
    create_access_token,
)
from app.gateway.auth.config import get_auth_config
from app.gateway.auth.errors import AuthErrorCode, AuthErrorResponse
from app.gateway.auth.models import User
from app.gateway.auth.password import has_account_password, hash_password_async, is_password_hash, verify_account_password
from app.gateway.auth.repositories.base import CredentialConflictError
from app.gateway.csrf_middleware import is_secure_request
from app.gateway.deps import get_agent_store, get_current_user_from_request, get_local_provider, get_run_store, get_skill_store
from app.gateway.perf_logging import finish_trace, trace_request, trace_stage
from deerflow.config import get_app_config
from deerflow.config.system_settings import AuthenticationSettings, load_system_settings
from deerflow.persistence.agents.base import AgentStore
from deerflow.persistence.skills.base import SkillStore

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/api/v1/auth", tags=["auth"])
_USER_NOT_LOOKED_UP = object()


# ── Request/Response Models ──────────────────────────────────────────────


class LoginResponse(BaseModel):
    """Response model for login — token only lives in HttpOnly cookie."""

    expires_in: int  # seconds
    needs_setup: bool = False


# Top common-password blocklist. Drawn from the public SecLists "10k worst
# passwords" set, lowercased + length>=8 only (shorter ones already fail
# the min_length check). Kept tight on purpose: this is the **lower bound**
# defense, not a full HIBP / passlib check, and runs in-process per request.
_COMMON_PASSWORDS: frozenset[str] = frozenset(
    {
        "password",
        "password1",
        "password12",
        "password123",
        "password1234",
        "12345678",
        "123456789",
        "1234567890",
        "qwerty12",
        "qwertyui",
        "qwerty123",
        "abc12345",
        "abcd1234",
        "iloveyou",
        "letmein1",
        "welcome1",
        "welcome123",
        "admin123",
        "administrator",
        "passw0rd",
        "p@ssw0rd",
        "monkey12",
        "trustno1",
        "sunshine",
        "princess",
        "football",
        "baseball",
        "superman",
        "batman123",
        "starwars",
        "dragon123",
        "master123",
        "shadow12",
        "michael1",
        "jennifer",
        "computer",
    }
)


def _password_is_common(password: str) -> bool:
    """Case-insensitive blocklist check.

    Lowercases the input so trivial mutations like ``Password`` /
    ``PASSWORD`` are also rejected. Does not normalize digit substitutions
    (``p@ssw0rd`` is included as a literal entry instead) — keeping the
    rule cheap and predictable.
    """
    return password.lower() in _COMMON_PASSWORDS


def _validate_strong_password(value: str) -> str:
    """Pydantic field-validator body shared by Register + ChangePassword.

    Constraint = function, not type-level mixin. The two request models
    have no "is-a" relationship; they only share the password-strength
    rule. Lifting it into a free function lets each model bind it via
    ``@field_validator(field_name)`` without inheritance gymnastics.
    """
    if _password_is_common(value):
        raise ValueError("密码过于常见，请设置更安全的密码")
    if value != value.strip():
        raise ValueError("密码首尾不能包含空格")
    return value


class RegisterRequest(BaseModel):
    """Request model for user registration."""

    username: str | None = Field(default=None, min_length=1, max_length=128)
    email: EmailStr | None = None
    password: str = Field(..., min_length=8, max_length=255)
    model_config = ConfigDict(extra="forbid")

    _strong_password = field_validator("password")(classmethod(lambda cls, v: _validate_strong_password(v)))

    @model_validator(mode="after")
    def validate_identity(self):
        if bool(self.username) == bool(self.email):
            raise ValueError("请填写用户名或邮箱（只需一项）")
        if self.username is not None:
            value = self.username.strip()
            if "@" not in value and (not value or value[0] in "._-" or value[-1] in "._-" or any(not (ch.isalnum() or ch in "._-") for ch in value)):
                raise ValueError("用户名仅支持字母、数字、点、下划线和短横线，首尾须为字母或数字")
            self.username = value
            if "@" in value:
                TypeAdapter(EmailStr).validate_python(value)
        return self


class UsernameLoginRequest(BaseModel):
    """Request model for passwordless username login.

    Behavior:
    - Existing user: direct login without password
    - Missing user: auto-register and login without password
    """

    username: str = Field(..., min_length=1, max_length=128)


class UsernameLoginResponse(BaseModel):
    """Response model for passwordless username login.

    Returns the access token directly in the body so non-browser clients
    can use it as a Bearer credential, while a session cookie is still set
    for browser-based callers.
    """

    access_token: str
    token_type: str = "bearer"
    expires_in: int
    user_id: str
    email: str
    system_role: Literal["admin", "user"] = "user"
    needs_setup: bool = False
    created: bool = Field(default=False, description="True when the user was auto-registered by this call")
    username: str | None = Field(
        default=None,
        description=(
            "The login username this call resolved (token login: the value read "
            "from getTokenInfo; username login: the supplied username). Used by "
            "the frontend to append ?username= on task-workspace jump links; the "
            "stored email local-part is lossy (lowercased/normalized)."
        ),
    )


class ChangePasswordRequest(BaseModel):
    """Request model for password change (also handles setup flow)."""

    current_password: str = Field(default="", max_length=255)
    new_password: str = Field(..., min_length=8, max_length=255)
    new_email: EmailStr | None = None

    _strong_password = field_validator("new_password")(classmethod(lambda cls, v: _validate_strong_password(v)))


class MessageResponse(BaseModel):
    """Generic message response."""

    message: str


class PasswordChangeResponse(MessageResponse):
    access_token: str
    token_type: str = "bearer"
    expires_in: int


# ── Helpers ───────────────────────────────────────────────────────────────


def _normalize_username_local_part(username: str) -> str | None:
    """Return the canonical local-part for a username, or None if user typed an email.

    Used both to build the synthetic email and to look up legacy accounts that
    were registered with a different default domain. Returning ``None`` for
    raw-email input tells callers to skip the domain-agnostic fallback —
    if you typed ``alice@old.com``, you meant that specific row.
    """
    raw = username.strip()
    if not raw:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="用户名不能为空")
    if "@" in raw:
        return None
    normalized = "".join(ch if (ch.isalnum() or ch in "._-") else "-" for ch in raw.lower()).strip("._-")
    if not normalized:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="用户名不合法")
    return normalized


def _username_to_email(username: str) -> str:
    """Map a frontend username to a valid email used by user storage.

    - If username already looks like an email (`contains @`), use it directly.
    - Otherwise generate `<normalized-username>@<domain>`.
    """
    local = _normalize_username_local_part(username)
    if local is None:
        return username.strip().lower()

    # Default mail domain for passwordless / username-only registrations.
    # Kept overridable via env so individual deployments can swap it out, but the
    # built-in default targets the cm.com tenant.
    domain = os.getenv("DEER_FLOW_USERNAME_LOGIN_DOMAIN", "cm.com").strip().lower() or "cm.com"
    return f"{local}@{domain}"


# 后台岗位同步任务的强引用集合：防止 create_task 的任务被 GC 中途回收。
_position_sync_tasks: set = set()
_POSITION_SYNC_CACHE_SECONDS = 60.0
_MAX_POSITION_SYNC_CACHE_USERS = 10_000
_position_sync_completed_at: dict[str, float] = {}
_position_sync_inflight: dict[str, asyncio.Task[None]] = {}


def _sync_user_position(request: Request, user_id: str) -> None:
    """Best-effort: schedule the user's 岗位 (position) sync off the login path.

    Ensures a user who has never been synced — e.g. a brand-new account that
    inherits the ``is_default`` position — gets the default agents / tasks /
    prompts loaded into their "我的" tables. The sync is idempotent but touches
    a dozen tables per user, so it runs as a fire-and-forget task instead of
    blocking the login response; the materialized rows land a moment later and
    failures are logged, never surfaced.
    """
    normalized_user_id = str(user_id)
    now = time.monotonic()
    completed_at = _position_sync_completed_at.get(normalized_user_id)
    if completed_at is not None and now - completed_at < _POSITION_SYNC_CACHE_SECONDS:
        return
    if normalized_user_id in _position_sync_inflight:
        return

    sync = getattr(request.app.state, "position_sync", None)
    if sync is None:
        return

    async def _run() -> None:
        try:
            await sync.sync_user(normalized_user_id)
            _position_sync_completed_at[normalized_user_id] = time.monotonic()
            if len(_position_sync_completed_at) > _MAX_POSITION_SYNC_CACHE_USERS:
                oldest = sorted(_position_sync_completed_at, key=_position_sync_completed_at.get)
                for stale_user_id in oldest[: len(oldest) // 2]:
                    _position_sync_completed_at.pop(stale_user_id, None)
        except Exception:
            logger.debug("position sync on login failed for user %s", user_id, exc_info=True)
        finally:
            _position_sync_inflight.pop(normalized_user_id, None)

    try:
        task = asyncio.create_task(_run())
    except RuntimeError:
        # 防御：请求处理器内必有运行中的事件循环，这里理论上走不到。
        logger.debug("position sync scheduling failed for user %s", user_id, exc_info=True)
        return
    _position_sync_tasks.add(task)
    _position_sync_inflight[normalized_user_id] = task
    task.add_done_callback(_position_sync_tasks.discard)


async def _get_or_create_user_by_username(
    username: str,
    *,
    existing_user: User | None | object = _USER_NOT_LOOKED_UP,
):
    """Resolve a username to a local user, auto-registering when absent.

    Shared by ``/login/username`` and ``/login/token`` — both turn an
    externally-supplied username into a local account and sign it in. Returns
    ``(user, created)`` where ``created`` is True only when this call inserted
    the row. Roles are persisted authorization data, never inferred from names.
    """
    provider = get_local_provider()
    email = _username_to_email(username)
    local_part = _normalize_username_local_part(username)

    created = False
    # A bare username is globally unique across synthetic email domains.  Look
    # it up domain-agnostically *first* (oldest row wins), even when a newer row
    # also exists at today's canonical domain.  Choosing the exact canonical
    # row first made the same username alternate backend user_ids after a
    # domain migration, which in turn made its existing threads inaccessible.
    if existing_user is not _USER_NOT_LOOKED_UP:
        # /login/username already performed this lookup to validate the
        # account-specific password. Reuse both hits and misses instead of
        # issuing the same MySQL query twice during a login burst.
        user = existing_user
    elif local_part is not None:
        user = await provider.find_user_by_email_local_part(local_part)
        if user is not None and user.email != email:
            logger.info("Matched username %r to stable legacy account %s", username, user.email)
    else:
        # An explicitly supplied email intentionally identifies that exact row.
        user = await provider.get_user_by_email(email)
    if user is None:
        try:
            # 黑名单模式:从未登录过的新用户(此处即首次自动注册)默认放行(approved=True)。
            # 管理员可在白名单管理页对个别用户「取消放行」来封禁——仅当登录白名单开关打开时生效。
            user = await provider.create_user(email=email, password=None, system_role="user", needs_setup=False, approved=True)
            created = True
            logger.info("Auto-registered passwordless user: %s", email)
        except ValueError:
            # Concurrent first login may create the same user in parallel.
            user = await provider.get_user_by_email(email)
            if user is None and local_part is not None:
                user = await provider.find_user_by_email_local_part(local_part)
            if user is None:
                raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="该用户名已存在，请使用原账号登录")
    return user, created


async def _lookup_user_by_username(username: str) -> User | None:
    """Resolve a username to an **existing** local user, without creating one.

    Mirrors the lookup half of ``_get_or_create_user_by_username`` (canonical
    ``<username>@<domain>`` email + legacy local-part fallback) so the
    per-account 登录口令 gate can read the target account's ``login_password``
    before deciding whether to admit the login. Returns ``None`` for a username
    that has never registered (the personal-password gate then simply does not
    apply, and ``_get_or_create_user_by_username`` later auto-registers it).
    """
    provider = get_local_provider()
    email = _username_to_email(username)
    local_part = _normalize_username_local_part(username)
    if local_part is not None:
        return await provider.find_user_by_email_local_part(local_part)
    return await provider.get_user_by_email(email)


def enforce_user_approved(user) -> None:
    """登录白名单拦截(**黑名单模式**):开关打开且用户被「取消放行」时拒绝登录。

    新账号首次登录即默认放行(``approved=True``,见 ``_get_or_create_user_by_username``
    与 ``create_user``),所以从未在本系统登录过的用户可以直接使用;白名单开关只用来
    封禁管理员显式「取消放行」(``approved=False``) 的个别用户。

    规则(三条硬约束防锁死):
    - 开关关(``login_whitelist.enabled`` 默认 False)→ 直接放行,人人可登录。
    - admin → 永远放行(避免管理员把自己关在门外)。
    - 其余:``approved`` 为 False(被管理员取消放行)→ 403 ``PENDING_APPROVAL``;
      前端据此提示「等待管理员开通」,且不写任何登录态。

    放在各登录端点拿到 user 之后、签发 JWT 之前调用。
    """
    try:
        enabled = load_system_settings().login_whitelist.enabled
    except Exception:
        # 读配置失败时绝不误伤(白名单是附加闸门,故障时回到「不拦截」)。
        logger.warning("读取登录白名单配置失败,本次登录不拦截", exc_info=True)
        return
    if not enabled:
        return
    if getattr(user, "system_role", None) == "admin":
        return
    if not getattr(user, "approved", False):
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail={"code": "PENDING_APPROVAL", "message": "您的账号已被限制登录,请联系管理员开通"},
        )


async def _fetch_username_from_token(token: str) -> str:
    """Resolve an upstream access token to a username via the configured endpoint.

    Calls ``auth_login.token_login.token_info_url`` with the caller-supplied
    token placed directly in the ``Authorization: Bearer <token>`` header — the
    upstream reads the bearer token from there, not from a cookie. Expects
    ``{"state": "200", "data": {"username": ...}}`` and returns ``data.username``.
    Any upstream failure, non-200 ``state``, or missing username raises HTTP 401.
    """
    config = get_app_config().auth_login.token_login

    # Offline test shortcut: a token listed in ``mock_tokens`` resolves to its
    # mapped username without ever calling the (possibly unreachable) upstream
    # ``token_info_url``. Lets token login be exercised end-to-end where the real
    # getTokenInfo host can't be reached (external networks). Empty in prod.
    mocked = config.mock_tokens.get(token)
    if mocked and str(mocked).strip():
        return str(mocked).strip()

    if not config.token_info_url:
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="token 登录未配置(token_info_url 为空)")

    # TLS verification for the (HTTPS) upstream. A custom CA bundle wins; else
    # the boolean toggle. UAT/internal hosts often need verify_ssl=false because
    # they serve self-signed or private-CA certs — without this the call dies
    # with an SSLError and every token login silently 401s.
    verify: bool | str = config.ca_cert_path if config.ca_cert_path else config.verify_ssl

    try:
        async with httpx.AsyncClient(timeout=config.timeout_seconds, verify=verify) as client:
            upstream = await client.post(
                config.token_info_url,
                headers={
                    "Authorization": f"Bearer {token}",
                    "Content-Type": "application/x-www-form-urlencoded",
                },
            )
    except httpx.HTTPError as exc:
        # Surface the concrete failure (SSL handshake, DNS, timeout, ...) in logs
        # so HTTPS/cert misconfiguration is diagnosable; client still sees 401.
        logger.warning(
            "Token login: upstream call to %s failed (verify=%r): %s: %s",
            config.token_info_url,
            verify,
            type(exc).__name__,
            exc,
        )
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=AuthErrorResponse(code=AuthErrorCode.INVALID_CREDENTIALS, message="token 校验失败").model_dump())

    if upstream.status_code != 200:
        logger.warning("Token login: upstream returned HTTP %s", upstream.status_code)
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=AuthErrorResponse(code=AuthErrorCode.INVALID_CREDENTIALS, message="token 无效").model_dump())

    try:
        payload = upstream.json()
    except ValueError:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=AuthErrorResponse(code=AuthErrorCode.INVALID_CREDENTIALS, message="token 接口返回格式异常").model_dump())

    # Upstream signals success with the string state "200" (see API docs).
    if str(payload.get("state")) != "200":
        logger.warning("Token login: upstream state=%r msg=%r", payload.get("state"), payload.get("msg"))
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=AuthErrorResponse(code=AuthErrorCode.INVALID_CREDENTIALS, message="token 无效或已过期").model_dump())

    username = (payload.get("data") or {}).get("username")
    if not username or not str(username).strip():
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=AuthErrorResponse(code=AuthErrorCode.INVALID_CREDENTIALS, message="未能从 token 解析出用户名").model_dump())

    return str(username).strip()


def _set_session_cookie(response: Response, token: str, request: Request) -> None:
    """Set the access_token HttpOnly cookie on the response."""
    config = get_auth_config()
    is_https = is_secure_request(request)
    response.set_cookie(
        key="access_token",
        value=token,
        httponly=True,
        secure=is_https,
        samesite="lax",
        max_age=config.token_expiry_days * 24 * 3600 if is_https else None,
    )


# ── Rate Limiting ────────────────────────────────────────────────────────
# In-process dict — not shared across workers.
#
# **Limitation**: with multi-worker deployments (e.g., gunicorn -w N), each
# worker maintains its own lockout table, so an attacker effectively gets
# N × _MAX_LOGIN_ATTEMPTS guesses before being locked out everywhere. For
# production multi-worker setups, replace this with a shared store (Redis,
# database-backed counter) to enforce a true per-IP limit.

_MAX_LOGIN_ATTEMPTS = 100
_LOCKOUT_SECONDS = 300  # 5 minutes

# ip → (fail_count, lock_until_timestamp)
_login_attempts: dict[str, tuple[int, float]] = {}


def _trusted_proxies() -> list:
    """Parse ``AUTH_TRUSTED_PROXIES`` env var into a list of ip_network objects.

    Comma-separated CIDR or single-IP entries. Empty / unset = no proxy is
    trusted (direct mode). Invalid entries are skipped with a logger warning.
    Read live so env-var overrides take effect immediately and tests can
    ``monkeypatch.setenv`` without poking a module-level cache.
    """
    raw = os.getenv("AUTH_TRUSTED_PROXIES", "").strip()
    if not raw:
        return []
    nets = []
    for entry in raw.split(","):
        entry = entry.strip()
        if not entry:
            continue
        try:
            nets.append(ip_network(entry, strict=False))
        except ValueError:
            logger.warning("AUTH_TRUSTED_PROXIES: ignoring invalid entry %r", entry)
    return nets


def _get_client_ip(request: Request) -> str:
    """Extract the real client IP for rate limiting.

    Trust model:

    - The TCP peer (``request.client.host``) is always the baseline. It is
      whatever the kernel reports as the connecting socket — unforgeable
      by the client itself.
    - ``X-Real-IP`` is **only** honored if the TCP peer is in the
      ``AUTH_TRUSTED_PROXIES`` allowlist (set via env var, comma-separated
      CIDR or single IPs). When set, the gateway is assumed to be behind a
      reverse proxy (nginx, Cloudflare, ALB, …) that overwrites
      ``X-Real-IP`` with the original client address.
    - With no ``AUTH_TRUSTED_PROXIES`` set, ``X-Real-IP`` is silently
      ignored — closing the bypass where any client could rotate the
      header to dodge per-IP rate limits in dev / direct-gateway mode.

    ``X-Forwarded-For`` is intentionally NOT used because it is naturally
    client-controlled at the *first* hop and the trust chain is harder to
    audit per-request.
    """
    peer_host = request.client.host if request.client else None

    trusted = _trusted_proxies()
    if trusted and peer_host:
        try:
            peer_ip = ip_address(peer_host)
            if any(peer_ip in net for net in trusted):
                real_ip = request.headers.get("x-real-ip", "").strip()
                if real_ip:
                    return real_ip
        except ValueError:
            # peer_host wasn't a parseable IP (e.g. "unknown") — fall through
            pass

    return peer_host or "unknown"


def _check_rate_limit(ip: str) -> None:
    """Raise 429 if the IP is currently locked out."""
    record = _login_attempts.get(ip)
    if record is None:
        return
    fail_count, lock_until = record
    if fail_count >= _MAX_LOGIN_ATTEMPTS:
        if time.time() < lock_until:
            raise HTTPException(
                status_code=429,
                detail="登录尝试次数过多,请稍后再试",
            )
        del _login_attempts[ip]


_MAX_TRACKED_IPS = 10000


def _record_login_failure(ip: str) -> None:
    """Record a failed login attempt for the given IP."""
    # Evict expired lockouts when dict grows too large
    if len(_login_attempts) >= _MAX_TRACKED_IPS:
        now = time.time()
        expired = [k for k, (c, t) in _login_attempts.items() if c >= _MAX_LOGIN_ATTEMPTS and now >= t]
        for k in expired:
            del _login_attempts[k]
        # If still too large, evict cheapest-to-lose half: below-threshold
        # IPs (lock_until=0.0) sort first, then earliest-expiring lockouts.
        if len(_login_attempts) >= _MAX_TRACKED_IPS:
            by_time = sorted(_login_attempts.items(), key=lambda kv: kv[1][1])
            for k, _ in by_time[: len(by_time) // 2]:
                del _login_attempts[k]

    record = _login_attempts.get(ip)
    if record is None:
        _login_attempts[ip] = (1, 0.0)
    else:
        new_count = record[0] + 1
        lock_until = time.time() + _LOCKOUT_SECONDS if new_count >= _MAX_LOGIN_ATTEMPTS else 0.0
        _login_attempts[ip] = (new_count, lock_until)


def _record_login_success(ip: str) -> None:
    """Clear failure counter for the given IP on successful login."""
    _login_attempts.pop(ip, None)


# ── Endpoints ─────────────────────────────────────────────────────────────


def _require_login_method(method: str) -> None:
    settings = load_system_settings().authentication
    if not getattr(settings, method):
        labels = {"password_login_enabled": "账号密码登录", "registration_enabled": "自助注册", "sso_enabled": "外部系统免登"}
        raise HTTPException(status_code=403, detail=f"管理员尚未开启{labels[method]}")


def _login_response(user: User, token: str, *, username: str | None = None, created: bool = False) -> UsernameLoginResponse:
    return UsernameLoginResponse(
        access_token=token,
        expires_in=get_auth_config().token_expiry_days * 24 * 3600,
        user_id=str(user.id),
        email=user.email,
        system_role=user.system_role,
        needs_setup=user.needs_setup,
        created=created,
        username=username or user.email.split("@", 1)[0],
    )


async def _prepare_login_session(user: User) -> User:
    """Apply the administrator's one-online-session policy after authentication."""
    if not load_system_settings().authentication.single_session_enabled:
        return user
    try:
        return await get_local_provider().start_login_session(user)
    except CredentialConflictError as exc:
        raise HTTPException(status_code=409, detail=str(exc)) from exc


class AuthenticationOptions(AuthenticationSettings):
    needs_setup: bool = False


@router.get("/options", response_model=AuthenticationOptions)
async def authentication_options(request: Request, response: Response) -> AuthenticationOptions:
    """Anonymous, non-secret capability discovery for the login page."""
    response.headers["Cache-Control"] = "no-store"
    trace = trace_request(request, "auth_options")
    try:
        stage_started = time.perf_counter()
        authentication = load_system_settings().authentication.model_dump()
        trace_stage(trace, "load_system_settings", stage_started)

        stage_started = time.perf_counter()
        needs_setup = await get_local_provider().count_admin_users() == 0
        trace_stage(trace, "count_admin_users", stage_started)
        return AuthenticationOptions(**authentication, needs_setup=needs_setup)
    finally:
        finish_trace(trace)


@router.post("/login/local", response_model=UsernameLoginResponse)
async def login_local(
    request: Request,
    response: Response,
    form_data: OAuth2PasswordRequestForm = Depends(),
):
    """Username or email login with the account's unified password/passphrase."""
    _require_login_method("password_login_enabled")
    client_ip = _get_client_ip(request)
    _check_rate_limit(client_ip)

    user = await _lookup_user_by_username(form_data.username)
    if user is not None and not await verify_account_password(form_data.password, user):
        user = None

    if user is None:
        _record_login_failure(client_ip)
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=AuthErrorResponse(code=AuthErrorCode.INVALID_CREDENTIALS, message="账号或密码错误").model_dump(),
        )

    _record_login_success(client_ip)
    enforce_user_approved(user)
    user = await _prepare_login_session(user)
    token = create_access_token(str(user.id), token_version=user.token_version, auth_method="password")
    _set_session_cookie(response, token, request)
    _sync_user_position(request, user.id)

    return _login_response(user, token, username=form_data.username.strip())


def _username_login_request(username: object) -> UsernameLoginRequest:
    if not isinstance(username, str):
        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="username is required")
    try:
        return UsernameLoginRequest(username=username)
    except ValidationError as exc:
        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="username is required") from exc


async def _read_username_login(request: Request) -> UsernameLoginRequest:
    """Accept the historical JSON body and a form body.

    ``application/x-www-form-urlencoded`` is a CORS simple content type. Old
    Firefox aborts a credentialed JSON login during preflight and reports
    NS_BINDING_ABORTED, so the browser client sends a form body instead.
    """
    media_type = request.headers.get("content-type", "").split(";", 1)[0].strip().lower()
    if media_type == "application/x-www-form-urlencoded":
        form = await request.form()
        return _username_login_request(form.get("username"))
    try:
        payload = await request.json()
    except json.JSONDecodeError as exc:
        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="username is required") from exc
    username = payload.get("username") if isinstance(payload, dict) else None
    return _username_login_request(username)


@router.post("/login/username", response_model=UsernameLoginResponse)
async def login_by_username(
    request: Request,
    response: Response,
    password: str | None = Query(default=None, description="Account password or shared passphrase; required when password login is enabled."),
):
    """Compatibility username login, gated by the current credential policy.

    Personal password/passphrase wins. With password login enabled, accounts
    without personal credentials must supply the existing shared passphrase
    (default 123ewq). Identity-only login is retained only while password login
    is disabled and the historical shared-password gate is also disabled.
    """
    body = await _read_username_login(request)
    _require_login_method("sso_enabled")
    username_login_config = get_app_config().auth_login.username_login
    password_login_enabled = load_system_settings().authentication.password_login_enabled

    # Per-account password takes precedence: look the account up (no create) and,
    # if it carries a personal 登录口令, require an exact match — ignoring the
    # shared gate entirely for that account.
    existing = await _lookup_user_by_username(body.username)
    credentials_verified = False
    if has_account_password(existing) or password_login_enabled or username_login_config.require_password:
        client_ip = _get_client_ip(request)
        _check_rate_limit(client_ip)
        if not await verify_account_password(password or "", existing):
            _record_login_failure(client_ip)
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail=AuthErrorResponse(code=AuthErrorCode.INVALID_CREDENTIALS, message="口令错误").model_dump(),
            )
        credentials_verified = True
        _record_login_success(client_ip)

    user, created = await _get_or_create_user_by_username(body.username, existing_user=existing)

    # A registration/reset may commit after the initial lookup/password gate.
    # Never mint the new account/version's session using an earlier decision.
    if has_account_password(user) and (existing is None or existing.id != user.id or existing.token_version != user.token_version):
        client_ip = _get_client_ip(request)
        _check_rate_limit(client_ip)
        if not await verify_account_password(password or "", user):
            _record_login_failure(client_ip)
            raise HTTPException(status_code=401, detail=AuthErrorResponse(code=AuthErrorCode.INVALID_CREDENTIALS, message="口令错误").model_dump())
        credentials_verified = True

    enforce_user_approved(user)
    user = await _prepare_login_session(user)
    token = create_access_token(str(user.id), token_version=user.token_version, auth_method="passphrase" if credentials_verified else None)
    _set_session_cookie(response, token, request)
    _sync_user_position(request, user.id)

    return UsernameLoginResponse(
        access_token=token,
        token_type="bearer",
        expires_in=get_auth_config().token_expiry_days * 24 * 3600,
        user_id=str(user.id),
        email=user.email,
        system_role=user.system_role,
        needs_setup=user.needs_setup,
        created=created,
        username=body.username,
    )


@router.post("/login/token", response_model=UsernameLoginResponse)
async def login_by_token(
    request: Request,
    response: Response,
    token: str = Query(..., min_length=1, description="Upstream access token to exchange for a username."),
):
    """Token-exchange login.

    Verifies an upstream access token by calling the configured
    ``token_info_url`` (see ``auth_login.token_login``), reads the resolved
    username, then signs the matching local account in — auto-registering it
    when the username is not yet in the database. Gated by
    ``auth_login.token_login.enabled``.

    Returns the JWT access token in the response body and also sets it as a
    session cookie for browser callers.
    """
    _require_login_method("sso_enabled")
    config = get_app_config().auth_login.token_login
    if not config.enabled:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="token 登录未开启")

    client_ip = _get_client_ip(request)
    _check_rate_limit(client_ip)

    try:
        username = await _fetch_username_from_token(token)
    except HTTPException as exc:
        if exc.status_code == status.HTTP_401_UNAUTHORIZED:
            _record_login_failure(client_ip)
        raise

    _record_login_success(client_ip)

    user, created = await _get_or_create_user_by_username(username)

    enforce_user_approved(user)
    user = await _prepare_login_session(user)
    access_token = create_access_token(str(user.id), token_version=user.token_version, auth_method="sso")
    _set_session_cookie(response, access_token, request)
    _sync_user_position(request, user.id)

    return UsernameLoginResponse(
        access_token=access_token,
        token_type="bearer",
        expires_in=get_auth_config().token_expiry_days * 24 * 3600,
        user_id=str(user.id),
        email=user.email,
        system_role=user.system_role,
        needs_setup=user.needs_setup,
        created=created,
        # The raw username resolved from getTokenInfo (pre email-normalization),
        # so the frontend can append the exact ?username= on jump links.
        username=username,
    )


@router.post("/register", response_model=UsernameLoginResponse, status_code=status.HTTP_201_CREATED)
async def register(request: Request, response: Response, body: RegisterRequest):
    """Register a new user account (always 'user' role).

    Admin is auto-created on first boot. This endpoint creates regular users.
    Auto-login by setting the session cookie.
    """
    _require_login_method("password_login_enabled")
    _require_login_method("registration_enabled")
    client_ip = _get_client_ip(request)
    _check_rate_limit(client_ip)
    provider = get_local_provider()
    username = body.username or str(body.email)
    email = _username_to_email(username)
    # Match every domain, even for explicit emails: a self-registration must
    # never shadow the stable account selected by bare-username login.
    duplicate = await provider.find_user_by_email_local_part(email.split("@", 1)[0])
    try:
        if duplicate is not None:
            raise ValueError("Duplicate username")
        user = await provider.create_user(email=email, password=body.password, system_role="user", approved=True)
    except ValueError:
        _record_login_failure(client_ip)
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=AuthErrorResponse(code=AuthErrorCode.EMAIL_ALREADY_EXISTS, message="该用户名或邮箱已注册").model_dump(),
        )

    enforce_user_approved(user)
    user = await _prepare_login_session(user)
    token = create_access_token(str(user.id), token_version=user.token_version, auth_method="password")
    _set_session_cookie(response, token, request)
    _sync_user_position(request, user.id)

    return _login_response(user, token, username=username, created=True)


@router.post("/logout", response_model=MessageResponse)
async def logout(request: Request, response: Response):
    """Logout current user by clearing the cookie."""
    response.delete_cookie(key="access_token", secure=is_secure_request(request), samesite="lax")
    return MessageResponse(message="Successfully logged out")


@router.post("/change-password", response_model=PasswordChangeResponse)
async def change_password(request: Request, response: Response, body: ChangePasswordRequest):
    """Change password for the currently authenticated user.

    Also handles the first-boot setup flow:
    - If new_email is provided, updates email (checks uniqueness)
    - If user.needs_setup is True and new_email is given, clears needs_setup
    - Always increments token_version to invalidate old sessions
    - Re-issues session cookie with new token_version
    """
    user = await get_current_user_from_request(request)
    client_ip = _get_client_ip(request)
    _check_rate_limit(client_ip)
    if not await verify_account_password(body.current_password, user):
        _record_login_failure(client_ip)
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=AuthErrorResponse(code=AuthErrorCode.INVALID_CREDENTIALS, message="当前密码错误").model_dump())

    provider = get_local_provider()

    # Update email if provided
    if body.new_email is not None:
        existing = await provider.find_user_by_email_local_part(body.new_email.split("@", 1)[0])
        if existing and str(existing.id) != str(user.id):
            raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=AuthErrorResponse(code=AuthErrorCode.EMAIL_ALREADY_EXISTS, message="Email already in use").model_dump())
    try:
        user = await provider.update_credentials(user, await hash_password_async(body.new_password), email=body.new_email)
    except ValueError as exc:
        raise HTTPException(status_code=409, detail="账号信息已变更，请刷新后重试") from exc
    _record_login_success(client_ip)

    # Re-issue cookie with new token_version
    token = create_access_token(str(user.id), token_version=user.token_version, auth_method="password")
    _set_session_cookie(response, token, request)

    return PasswordChangeResponse(message="密码已更新", access_token=token, expires_in=get_auth_config().token_expiry_days * 24 * 3600)


@router.get("/me", response_model=UserResponse)
async def get_me(request: Request):
    """Get current authenticated user info."""
    user = await get_current_user_from_request(request)
    return UserResponse(
        id=str(user.id),
        email=user.email,
        system_role=user.system_role,
        needs_setup=user.needs_setup,
        has_password=has_account_password(user),
        requires_current_password=True,
        uses_default_password=not has_account_password(user),
    )


_SETUP_STATUS_COOLDOWN: dict[str, float] = {}
_SETUP_STATUS_COOLDOWN_SECONDS = 60
_MAX_TRACKED_SETUP_STATUS_IPS = 10000


@router.get("/setup-status")
async def setup_status(request: Request):
    """Check if an admin account exists. Returns needs_setup=True when no admin exists."""
    client_ip = _get_client_ip(request)
    now = time.time()
    last_check = _SETUP_STATUS_COOLDOWN.get(client_ip, 0)
    elapsed = now - last_check
    if elapsed < _SETUP_STATUS_COOLDOWN_SECONDS:
        retry_after = max(1, int(_SETUP_STATUS_COOLDOWN_SECONDS - elapsed))
        raise HTTPException(
            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
            detail="Setup status check is rate limited",
            headers={"Retry-After": str(retry_after)},
        )
    # Evict stale entries when dict grows too large to bound memory usage.
    if len(_SETUP_STATUS_COOLDOWN) >= _MAX_TRACKED_SETUP_STATUS_IPS:
        cutoff = now - _SETUP_STATUS_COOLDOWN_SECONDS
        stale = [k for k, t in _SETUP_STATUS_COOLDOWN.items() if t < cutoff]
        for k in stale:
            del _SETUP_STATUS_COOLDOWN[k]
        # If still too large after evicting expired entries, remove oldest half.
        if len(_SETUP_STATUS_COOLDOWN) >= _MAX_TRACKED_SETUP_STATUS_IPS:
            by_time = sorted(_SETUP_STATUS_COOLDOWN.items(), key=lambda kv: kv[1])
            for k, _ in by_time[: len(by_time) // 2]:
                del _SETUP_STATUS_COOLDOWN[k]
    _SETUP_STATUS_COOLDOWN[client_ip] = now
    admin_count = await get_local_provider().count_admin_users()
    return {"needs_setup": admin_count == 0}


class InitializeAdminRequest(BaseModel):
    """Request model for first-boot admin account creation."""

    email: EmailStr
    password: str = Field(..., min_length=8, max_length=255)

    _strong_password = field_validator("password")(classmethod(lambda cls, v: _validate_strong_password(v)))


@router.post("/initialize", response_model=UsernameLoginResponse, status_code=status.HTTP_201_CREATED)
async def initialize_admin(request: Request, response: Response, body: InitializeAdminRequest):
    """Create the first admin account on initial system setup.

    Only callable when no admin exists. Returns 409 Conflict if an admin
    already exists.

    On success, the admin account is created with ``needs_setup=False`` and
    the session cookie is set.
    """
    admin_count = await get_local_provider().count_admin_users()
    if admin_count > 0:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=AuthErrorResponse(code=AuthErrorCode.SYSTEM_ALREADY_INITIALIZED, message="System already initialized").model_dump(),
        )

    try:
        user = await get_local_provider().create_initial_admin(email=str(body.email), password=body.password)
    except ValueError:
        # DB unique-constraint race: another concurrent request beat us.
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=AuthErrorResponse(code=AuthErrorCode.SYSTEM_ALREADY_INITIALIZED, message="System already initialized").model_dump(),
        )

    user = await _prepare_login_session(user)
    token = create_access_token(str(user.id), token_version=user.token_version, auth_method="password")
    _set_session_cookie(response, token, request)

    return _login_response(user, token, created=True)


# ── Admin Endpoints ───────────────────────────────────────────────────────


class UserListItem(BaseModel):
    id: str
    email: str
    system_role: Literal["admin", "user"]
    skill_count: int
    agent_count: int
    qa_count: int = 0  # 问答次数: number of runs (one run = one question)
    approved: bool = True  # 登录白名单放行标记(白名单管理页据此放行/取消)
    login_password: str | None = None  # Legacy plaintext only; never return hashes.
    has_login_password: bool = False
    access_token: str | None = None
    token_type: str = "bearer"
    expires_in: int | None = None


def _legacy_login_password(user: User) -> str | None:
    value = user.login_password
    return value if value and not is_password_hash(value) else None


class UserListPage(BaseModel):
    """Paged user list — returned when the caller passes ``limit``."""

    items: list[UserListItem]
    total: int
    approved_count: int  # approved 或 admin（搜索范围内）
    limit: int
    offset: int


_USER_LIST_TIMEOUT_SECONDS = 15


def _user_list_skill_store(request: Request, include_counts: bool = True) -> SkillStore | None:
    return get_skill_store(request) if include_counts else None


def _user_list_agent_store(request: Request, include_counts: bool = True) -> AgentStore | None:
    return get_agent_store(request) if include_counts else None


@router.get("/users", response_model=list[UserListItem] | UserListPage)
async def list_all_users(
    request: Request,
    include_counts: bool = Query(True, description="False=只查 users 表，跳过 runs 统计与 skill/agent 计数（白名单管理页用，避免慢查询）"),
    q: str | None = Query(None, max_length=100, description="可选：按邮箱/用户ID 子串过滤（仅分页模式生效）"),
    limit: int | None = Query(None, ge=1, le=200, description="传入即分页返回 UserListPage；不传保持旧的全量列表"),
    offset: int = Query(0, ge=0),
    skill_store: SkillStore | None = Depends(_user_list_skill_store),
    agent_store: AgentStore | None = Depends(_user_list_agent_store),
):
    """List all users with their skill, agent and Q&A counts. Admin only.

    Passing ``limit`` switches to a paged response that reads one page of the
    users table plus two COUNT queries; omitting it keeps the legacy flat list.
    """
    try:
        async with asyncio.timeout(_USER_LIST_TIMEOUT_SECONDS if not include_counts else None):
            user = await get_current_user_from_request(request)
            if user.system_role != "admin":
                raise HTTPException(status_code=403, detail="Admin only")
            provider = get_local_provider()
            users = (
                await provider.list_users()
                if limit is None
                else await provider.list_users_page(q=q, limit=limit, offset=offset)
            )
            qa_counts: dict[str, int] = {}
            # One recent-window GROUP BY for everyone's 问答次数. This keeps the admin
            # user list from scanning the complete lifetime of the runs table.
            if include_counts:
                try:
                    qa_counts = await get_run_store(request).count_by_user(
                        since=recent_admin_query_cutoff()
                    )
                except Exception:
                    logger.warning("Failed to load per-user run counts for user list", exc_info=True)
            result = []
            for u in users:
                uid = str(u.id)
                result.append(
                    UserListItem(
                        id=uid,
                        email=u.email,
                        system_role=u.system_role,
                        skill_count=(await skill_store.count_owned(uid)) if include_counts else 0,
                        agent_count=(await agent_store.count_owned(uid)) if include_counts else 0,
                        qa_count=qa_counts.get(uid, 0),
                        approved=getattr(u, "approved", True),
                        login_password=_legacy_login_password(u),
                        has_login_password=has_account_password(u),
                    )
                )
            if limit is None:
                return result
            total = await provider.count_users_filtered(q=q)
            approved_count = await provider.count_users_filtered(q=q, approved_or_admin=True)
            return UserListPage(items=result, total=total, approved_count=approved_count, limit=limit, offset=offset)
    except TimeoutError as exc:
        logger.warning("User list timed out waiting for account data")
        raise HTTPException(status_code=503, detail="用户基础信息查询超时，请稍后重试") from exc


class UserApprovalRequest(BaseModel):
    approved: bool = Field(..., description="True=放行该用户登录;False=取消放行(下次登录被拦)")


@router.put("/users/{user_id}/approval", response_model=UserListItem)
async def set_user_approval(user_id: str, body: UserApprovalRequest, request: Request) -> UserListItem:
    """放行 / 取消放行某个注册用户(登录白名单)。Admin only。

    取消放行只拦该用户的下次登录(不动 ``token_version``,现有会话不强制下线)。
    """
    admin = await get_current_user_from_request(request)
    if admin.system_role != "admin":
        raise HTTPException(status_code=403, detail="Admin only")

    provider = get_local_provider()
    target = await provider.get_user(user_id)
    if target is None:
        raise HTTPException(status_code=404, detail="用户不存在")

    target.approved = bool(body.approved)
    try:
        saved = await provider.update_user(target)
    except CredentialConflictError as exc:
        raise HTTPException(status_code=409, detail=str(exc)) from exc
    return UserListItem(
        id=str(saved.id),
        email=saved.email,
        system_role=saved.system_role,
        skill_count=0,
        agent_count=0,
        qa_count=0,
        approved=saved.approved,
        login_password=_legacy_login_password(saved),
        has_login_password=has_account_password(saved),
    )


class UserLoginPasswordRequest(BaseModel):
    password: str | None = Field(
        default=None,
        max_length=255,
        description="设置账号密码（与登录口令一致）；不允许清除后退回免密码登录。",
    )


@router.put("/users/{user_id}/login-password", response_model=UserListItem)
@router.put("/users/{user_id}/reset-password", response_model=UserListItem)
async def set_user_login_password(user_id: str, body: UserLoginPasswordRequest, request: Request, response: Response) -> UserListItem:
    """Admin password reset; rotates both credential columns and all sessions."""
    admin = await get_current_user_from_request(request)
    if admin.system_role != "admin":
        raise HTTPException(status_code=403, detail="Admin only")

    provider = get_local_provider()
    target = await provider.get_user(user_id)
    if target is None:
        raise HTTPException(status_code=404, detail="用户不存在")

    pwd = body.password or ""
    if not pwd:
        raise HTTPException(status_code=400, detail="不能清除账号密码，请设置新密码")
    if len(pwd) < 8:
        raise HTTPException(status_code=400, detail="密码至少需要 8 个字符")
    try:
        _validate_strong_password(pwd)
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    try:
        saved = await provider.update_credentials(target, await hash_password_async(pwd))
    except CredentialConflictError as exc:
        raise HTTPException(status_code=409, detail=str(exc)) from exc
    token = None
    expires_in = None
    if str(admin.id) == str(saved.id):
        token = create_access_token(str(saved.id), token_version=saved.token_version, auth_method="password")
        _set_session_cookie(response, token, request)
        expires_in = get_auth_config().token_expiry_days * 24 * 3600
    return UserListItem(
        id=str(saved.id),
        email=saved.email,
        system_role=saved.system_role,
        skill_count=0,
        agent_count=0,
        qa_count=0,
        approved=saved.approved,
        login_password=None,
        has_login_password=True,
        access_token=token,
        expires_in=expires_in,
    )


# ── OAuth Endpoints (Future/Placeholder) ─────────────────────────────────


@router.get("/oauth/{provider}")
async def oauth_login(provider: str):
    """Initiate OAuth login flow.

    Redirects to the OAuth provider's authorization URL.
    Currently a placeholder - requires OAuth provider implementation.
    """
    _require_login_method("sso_enabled")
    if provider not in ["github", "google"]:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=f"Unsupported OAuth provider: {provider}",
        )

    raise HTTPException(
        status_code=status.HTTP_501_NOT_IMPLEMENTED,
        detail="OAuth login not yet implemented",
    )


@router.get("/callback/{provider}")
async def oauth_callback(provider: str, code: str, state: str):
    """OAuth callback endpoint.

    Handles the OAuth provider's callback after user authorization.
    Currently a placeholder.
    """
    _require_login_method("sso_enabled")
    raise HTTPException(
        status_code=status.HTTP_501_NOT_IMPLEMENTED,
        detail="OAuth callback not yet implemented",
    )