kite/backend/api/admin/users.py

from __future__ import annotations

import time
import uuid
from typing import Optional

from config import settings
from fastapi import APIRouter, Depends, HTTPException, Request, status
from pydantic import BaseModel, EmailStr, Field, field_validator
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from db.models import CloudConnection, Document, Quota, RefreshToken, Topic, User
from deps.auth import get_current_admin
from deps.db import get_db
from deps.utils import get_client_ip
from services.audit import write_audit_log
from services.auth import hash_password, revoke_all_refresh_tokens, validate_password_strength, verify_password
from storage import get_storage_backend, get_storage_backend_for_document
from api.admin.shared import _user_to_dict

router = APIRouter()  # NO prefix — parent __init__.py carries /api/admin (D-04)

# ── Constants ─────────────────────────────────────────────────────────────────

_DEFAULT_QUOTA_BYTES = 104857600  # 100 MB free-tier default (D-06)


# ── Request models ────────────────────────────────────────────────────────────

class UserCreate(BaseModel):
    handle: str
    email: EmailStr
    password: str
    role: str = "user"

    @field_validator("password")
    @classmethod
    def password_strength(cls, v: str) -> str:
        validate_password_strength(v)
        return v


class UserStatusUpdate(BaseModel):
    is_active: bool


class UserAiConfigUpdate(BaseModel):
    ai_provider: Optional[str] = None
    ai_model: Optional[str] = None


class SystemTopicCreate(BaseModel):
    """Request model for admin system topic creation (D-09)."""

    name: str
    description: str = ""
    color: str = "#6366f1"


class UserDeleteConfirm(BaseModel):
    """Admin password confirmation required before hard-deleting a user (ADMIN-02, T-05-11-01)."""

    admin_password: str = Field(..., min_length=1)


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

@router.get("/users")
async def list_users(
    session: AsyncSession = Depends(get_db),
    _admin: User = Depends(get_current_admin),
) -> dict:
    result = await session.execute(
        select(User).order_by(User.created_at.desc())
    )
    users = result.scalars().all()
    return {"items": [_user_to_dict(u) for u in users]}


@router.post("/users", status_code=status.HTTP_201_CREATED)
async def create_user(
    request: Request,
    body: UserCreate,
    session: AsyncSession = Depends(get_db),
    _admin: User = Depends(get_current_admin),
) -> dict:
    existing_email = await session.execute(
        select(User).where(User.email == str(body.email))
    )
    if existing_email.scalar_one_or_none() is not None:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="Email already registered",
        )

    existing_handle = await session.execute(
        select(User).where(User.handle == body.handle)
    )
    if existing_handle.scalar_one_or_none() is not None:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="Handle already taken",
        )

    new_user = User(
        id=uuid.uuid4(),
        handle=body.handle,
        email=str(body.email),
        password_hash=hash_password(body.password),
        role=body.role,
        is_active=True,
        totp_enabled=False,
        password_must_change=True,  # ADMIN-01: force password change on first login
    )
    session.add(new_user)

    quota = Quota(
        user_id=new_user.id,
        limit_bytes=_DEFAULT_QUOTA_BYTES,
        used_bytes=0,
    )
    session.add(quota)
    await session.flush()  # persist User + Quota before audit_log FK references them
    # D-13: admin user created event
    _ip_addr = get_client_ip(request)
    await write_audit_log(
        session,
        event_type="admin.user_created",
        user_id=new_user.id,
        actor_id=_admin.id,
        resource_id=new_user.id,
        ip_address=_ip_addr,
    )
    await session.commit()

    return {
        "id": str(new_user.id),
        "handle": new_user.handle,
        "email": new_user.email,
        "role": new_user.role,
        "created_at": new_user.created_at.isoformat() if new_user.created_at else None,
    }


@router.patch("/users/{user_id}/status")
async def update_user_status(
    user_id: uuid.UUID,
    body: UserStatusUpdate,
    request: Request,
    session: AsyncSession = Depends(get_db),
    _admin: User = Depends(get_current_admin),
) -> dict:
    user = await session.get(User, user_id)
    if user is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")

    # Guard: cannot deactivate the only remaining active admin (T-02-29)
    if not body.is_active and user.role == "admin":
        count_result = await session.execute(
            select(func.count(User.id)).where(
                User.role == "admin",
                User.is_active.is_(True),
            )
        )
        active_admin_count = count_result.scalar_one()
        if active_admin_count <= 1:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="Cannot deactivate the only admin",
            )

    _ip_addr = get_client_ip(request)
    user.is_active = body.is_active

    if not body.is_active:
        # Revoke all refresh tokens on deactivation
        await revoke_all_refresh_tokens(session, user.id)
        # Revoke any pre-deactivation access tokens still within their TTL (T-7.2-01)
        await request.app.state.redis.set(
            f"user_nbf:{user.id}",
            int(time.time()),
            ex=settings.access_token_expire_minutes * 60,
        )

    session.add(user)

    # D-13: user deactivated/activated event
    _event = "admin.user_deactivated" if not body.is_active else "admin.user_activated"
    await write_audit_log(
        session,
        event_type=_event,
        user_id=user.id,
        actor_id=_admin.id,
        resource_id=user.id,
        ip_address=_ip_addr,
    )
    await session.commit()

    return {
        "id": str(user.id),
        "handle": user.handle,
        "email": user.email,
        "is_active": user.is_active,
    }


@router.post("/users/{user_id}/password-reset", status_code=status.HTTP_202_ACCEPTED)
async def initiate_password_reset(
    user_id: uuid.UUID,
    session: AsyncSession = Depends(get_db),
    _admin: User = Depends(get_current_admin),
) -> dict:
    user = await session.get(User, user_id)
    if user is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")

    from services.auth import create_password_reset_token  # noqa: PLC0415
    from config import settings as _settings  # noqa: PLC0415

    reset_token = create_password_reset_token(str(user.id))
    reset_link = f"{_settings.frontend_url}/password-reset/confirm?token={reset_token}"

    # Deferred import to avoid circular imports (same pattern as document_tasks)
    from tasks.email_tasks import send_reset_email  # noqa: PLC0415
    send_reset_email.delay(user.email, reset_link)

    return {"message": "Password reset email sent"}


@router.patch("/users/{user_id}/ai-config")
async def update_ai_config(
    user_id: uuid.UUID,
    body: UserAiConfigUpdate,
    request: Request,
    session: AsyncSession = Depends(get_db),
    _admin: User = Depends(get_current_admin),
) -> dict:
    user = await session.get(User, user_id)
    if user is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")

    _ip_addr = get_client_ip(request)
    user.ai_provider = body.ai_provider
    user.ai_model = body.ai_model
    session.add(user)

    # D-13: AI provider assigned event
    await write_audit_log(
        session,
        event_type="admin.ai_provider_assigned",
        user_id=user_id,
        actor_id=_admin.id,
        resource_id=None,
        ip_address=_ip_addr,
        metadata_={"provider": body.ai_provider, "model": body.ai_model},
    )
    await session.commit()

    return {
        "id": str(user.id),
        "email": user.email,
        "ai_provider": user.ai_provider,
        "ai_model": user.ai_model,
    }


@router.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_user(
    user_id: uuid.UUID,
    body: UserDeleteConfirm,
    request: Request,
    session: AsyncSession = Depends(get_db),
    _admin: User = Depends(get_current_admin),
) -> None:
    # T-05-11-01: Verify admin password before performing any destructive action.
    # Fail fast — no DB reads for the target user until the admin is confirmed.
    if not verify_password(body.admin_password, _admin.password_hash):
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Invalid admin password",
        )

    user = await session.get(User, user_id)
    if user is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")

    # T-04-07-04: Cannot delete admin accounts
    if user.role == "admin":
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Cannot delete admin accounts",
        )

    _ip_addr = get_client_ip(request)

    # SEC-09 (cloud): purge cloud-stored documents and credentials BEFORE DB delete.
    # Must run before MinIO cleanup so that credentials are still available to build
    # the cloud backend instances for delete_object calls.
    cloud_conns_result = await session.execute(
        select(CloudConnection).where(CloudConnection.user_id == user_id)
    )
    cloud_conns = cloud_conns_result.scalars().all()
    for conn in cloud_conns:
        # Delete cloud objects stored in this provider for this user
        cloud_docs_result = await session.execute(
            select(Document).where(
                Document.user_id == user_id,
                Document.storage_backend == conn.provider,
            )
        )
        for doc in cloud_docs_result.scalars().all():
            try:
                backend = await get_storage_backend_for_document(doc, user, session)
                await backend.delete_object(doc.object_key)
            except Exception:
                pass  # Best-effort cloud object cleanup; deletion proceeds regardless
        # Purge the credentials row (FK cascade would also remove it, but explicit
        # deletion here guarantees credentials_enc is gone before commit — SEC-09)
        await session.delete(conn)
    if cloud_conns:
        await session.flush()  # Flush connection deletes before user delete
        await write_audit_log(
            session,
            event_type="cloud.credentials_purged",
            user_id=user_id,
            actor_id=_admin.id,
            resource_id=user_id,
            ip_address=_ip_addr,
            metadata_={"providers": [c.provider for c in cloud_conns]},
        )

    # SEC-09 (minio): collect all user documents and delete MinIO objects BEFORE DB delete
    docs_result = await session.execute(
        select(Document).where(Document.user_id == user_id)
    )
    user_docs = docs_result.scalars().all()

    storage = get_storage_backend()
    for doc in user_docs:
        try:
            await storage.delete_object(doc.object_key)
        except Exception:
            pass  # Best-effort MinIO cleanup; DB deletion proceeds regardless

    # D-13: audit log BEFORE deleting the user row (user FK still valid at flush time)
    await write_audit_log(
        session,
        event_type="admin.user_deleted",
        user_id=user_id,
        actor_id=_admin.id,
        resource_id=user_id,
        ip_address=_ip_addr,
    )
    await session.flush()

    # Delete user record (CASCADE removes quota, documents, refresh_tokens, etc.)
    await session.delete(user)
    await session.commit()


@router.post("/topics", status_code=status.HTTP_201_CREATED)
async def create_system_topic(
    body: SystemTopicCreate,
    session: AsyncSession = Depends(get_db),
    _admin: User = Depends(get_current_admin),
) -> dict:
    from services import storage  # noqa: PLC0415

    topic = await storage.create_topic(
        session, body.name, body.description, body.color, user_id=None
    )
    return topic