ai-agent/backend/users/routes.py

"""Built-in login, ``/api/me``, the owner's user admin and conversation sharing."""

from __future__ import annotations

import time

from fastapi import APIRouter, HTTPException, Request, Response
from pydantic import BaseModel

import schemas
from core.http import _r
from core.state import State
from projects.catalog import ROOT_SLUG
from users import access, mailer
from users.caller import (OWNER_NAME, SESSION_COOKIE, CallerDep, OwnerDep,
                          guest_from_record, on_guest_surface, request_host)
from users.deliver import GUEST_PUBLIC_URL
from users.store import (DEFAULT_MODELS, INVITE_TTL_S, LEVELS, LINK_TTL_S,
                         NOTIFY_CHANNELS,
                         SESSION_TTL_S, norm_email, valid_email)

router = APIRouter()

# One login mail per address per this many seconds (the endpoint is public).
_LINK_COOLDOWN_S = 60


def _origin(request: Request) -> str:
    """Where a login link should land: the guest surface's public URL, else
    the host this request came in on."""
    if GUEST_PUBLIC_URL:
        return GUEST_PUBLIC_URL
    proto = request.headers.get("x-forwarded-proto") or request.url.scheme
    return f"{proto}://{request.headers.get('host') or request_host(request)}"


def _secure(request: Request) -> bool:
    return (request.headers.get("x-forwarded-proto") or request.url.scheme) == "https"


def _me(deps, caller) -> dict:
    out = {"id": caller.id, "role": caller.role, "name": caller.name,
           "email": caller.email, "readonly": caller.readonly,
           "projects": [{"slug": s, "level": l} for s, l in sorted(caller.projects.items())],
           "models": list(caller.models), "budgetUsd": caller.budget_usd,
           "notify": None, "spentUsd": None, "mail": mailer.enabled()}
    if not caller.is_owner:
        u = deps.users.get(caller.id) or {}
        out["notify"] = u.get("notify") or {"channels": ["in_app"]}
        out["defaultShortcuts"] = deps.users.defaults()["promptShortcuts"]
        from conversations.catalog import _conversation_cards
        from users.runs import month_spend
        out["spentUsd"] = round(month_spend(
            deps, caller.id, list(_conversation_cards(deps, False))), 4)
    return out


# ── login ────────────────────────────────────────────────────────────────────
class LinkRequest(BaseModel):
    email: str


@router.post("/api/auth/request-link", responses=_r(schemas.OkResponse))
def request_link(deps: State, request: Request, body: LinkRequest):
    """Mail a one-time sign-in link. Always answers ok — whether the address
    is a user is nobody's business but the owner's."""
    u = deps.users.by_email(body.email)
    if u and not u.get("disabled") and mailer.enabled():
        if time.time() - deps.auth.last_link_at(u["id"]) >= _LINK_COOLDOWN_S:
            token = deps.auth.mint_link(u["id"])
            try:
                mailer.login_mail(u["email"], u.get("name") or u["id"],
                                  f"{_origin(request)}/login?token={token}",
                                  LINK_TTL_S // 60)
            except mailer.MailError:
                pass
    return {"ok": True}


class LoginBody(BaseModel):
    token: str


@router.post("/api/auth/login", responses=_r(schemas.Me))
def login(deps: State, request: Request, response: Response, body: LoginBody):
    uid = deps.auth.consume_link(body.token)
    u = deps.users.active(uid) if uid else None
    if not u:
        raise HTTPException(401, {"message": "this link is invalid or expired",
                                  "kind": "link_invalid"})
    # The session this browser already had (if any) gives way to the new one.
    deps.auth.close_session(request.cookies.get(SESSION_COOKIE, ""))
    token = deps.auth.open_session(u["id"], request.headers.get("user-agent", ""))
    deps.users.touch_login(u["id"])
    response.set_cookie(SESSION_COOKIE, token, max_age=SESSION_TTL_S,
                        httponly=True, secure=_secure(request), samesite="lax",
                        path="/")
    return _me(deps, guest_from_record(u))


@router.post("/api/auth/logout", responses=_r(schemas.OkResponse))
def logout(deps: State, request: Request, response: Response):
    deps.auth.close_session(request.cookies.get(SESSION_COOKIE, ""))
    response.delete_cookie(SESSION_COOKIE, path="/")
    return {"ok": True}


# ── me ───────────────────────────────────────────────────────────────────────
@router.get("/api/me", responses=_r(schemas.Me))
def me(deps: State, request: Request, caller: CallerDep):
    return {**_me(deps, caller), "surface": "guest" if on_guest_surface(request)
            else "admin"}


class MePatch(BaseModel):
    name: str | None = None
    notify: dict | None = None


@router.put("/api/me", responses=_r(schemas.Me))
def me_update(deps: State, request: Request, caller: CallerDep, body: MePatch):
    """A guest's own preferences: display name and notification channels."""
    if caller.is_owner:
        raise HTTPException(400, "the owner's settings live in Settings")
    deps.users.update(caller.id, body.model_dump(exclude_none=True))
    u = deps.users.get(caller.id)
    return {**_me(deps, guest_from_record(u)), "surface": "guest"
            if on_guest_surface(request) else "admin"}


# ── user admin (owner) ───────────────────────────────────────────────────────
def _public(deps, u: dict) -> dict:
    return {**u, "sessions": len(deps.auth.sessions_of(u["id"]))}


@router.get("/api/users", responses=_r(schemas.UsersResponse))
def users_list(deps: State, _: OwnerDep):
    return {"users": [_public(deps, u) for u in
                      sorted(deps.users.all(), key=lambda u: u.get("name") or "")],
            "levels": list(LEVELS), "channels": list(NOTIFY_CHANNELS),
            "defaultModels": list(DEFAULT_MODELS), "mail": mailer.enabled(),
            "guestUrl": GUEST_PUBLIC_URL or None}


class UserBody(BaseModel):
    email: str | None = None
    name: str | None = None
    projects: dict | None = None
    models: list[str] | None = None
    budgetUsd: float | None = None
    notify: dict | None = None
    disabled: bool | None = None


def _check_grants(projects: dict | None) -> None:
    if projects and ROOT_SLUG in projects:
        raise HTTPException(400, f"the {ROOT_SLUG!r} repo can't be granted to a guest")


@router.post("/api/users", responses=_r(schemas.GuestUser))
def users_create(deps: State, _: OwnerDep, body: UserBody):
    if not body.email or not valid_email(body.email):
        raise HTTPException(400, "a valid email is required")
    _check_grants(body.projects)
    fields = body.model_dump(exclude_none=True)
    fields.pop("email", None)
    try:
        u = deps.users.create(norm_email(body.email), fields.pop("name", None), **fields)
    except ValueError as e:
        raise HTTPException(400, str(e))
    deps.hub.publish({"type": "users"})
    return _public(deps, u)


@router.put("/api/users/{uid}", responses=_r(schemas.GuestUser))
def users_update(deps: State, _: OwnerDep, uid: str, body: UserBody):
    _check_grants(body.projects)
    try:
        u = deps.users.update(uid, body.model_dump(exclude_none=True))
    except KeyError:
        raise HTTPException(404, "no such user")
    except ValueError as e:
        raise HTTPException(400, str(e))
    if u.get("disabled"):
        deps.auth.revoke_user(uid)
    deps.hub.publish({"type": "users"})
    return _public(deps, u)


@router.delete("/api/users/{uid}", responses=_r(schemas.OkResponse))
def users_delete(deps: State, _: OwnerDep, uid: str):
    if not deps.users.delete(uid):
        raise HTTPException(404, "no such user")
    deps.auth.revoke_user(uid)
    deps.hub.publish({"type": "users"})
    return {"ok": True}


class LinkBody(BaseModel):
    send: bool = False


@router.post("/api/users/{uid}/login-link", responses=_r(schemas.LoginLink))
def users_login_link(deps: State, request: Request, _: OwnerDep, uid: str,
                     body: LinkBody):
    """An invitation link — to copy into a message, or mailed (``send``) with
    an invitation note. Lives ``INVITE_TTL_S`` (days), unlike the quarter-hour
    link a guest asks for on the login page; still single-use."""
    u = deps.users.active(uid)
    if not u:
        raise HTTPException(404, "no such (enabled) user")
    url = f"{_origin(request)}/login?token={deps.auth.mint_link(uid, INVITE_TTL_S)}"
    sent = False
    if body.send:
        try:
            mailer.invite_mail(
                u["email"], u.get("name") or uid, url, inviter=OWNER_NAME,
                projects=sorted((u.get("projects") or {}).keys()),
                expires=_human_ttl(INVITE_TTL_S),
                home=f"{GUEST_PUBLIC_URL}/login" if GUEST_PUBLIC_URL else None)
            sent = True
        except mailer.MailError as e:
            raise HTTPException(502, str(e))
    return {"url": url, "sent": sent, "expiresInS": INVITE_TTL_S}


def _human_ttl(s: int) -> str:
    if s >= 2 * 86400:
        return f"{s // 86400} days"
    if s >= 2 * 3600:
        return f"{s // 3600} hours"
    return f"{max(1, s // 60)} minutes"


@router.delete("/api/users/{uid}/sessions")
def users_revoke(deps: State, _: OwnerDep, uid: str):
    return {"revoked": deps.auth.revoke_user(uid)}


# ── what every guest gets ────────────────────────────────────────────────────
class GuestDefaultsBody(BaseModel):
    promptShortcuts: list[dict] | None = None
    sharedSkills: list[str] | None = None


@router.get("/api/guest-defaults", responses=_r(schemas.GuestDefaults))
def guest_defaults_get(deps: State, _: OwnerDep):
    return deps.users.defaults()


@router.put("/api/guest-defaults", responses=_r(schemas.GuestDefaults))
def guest_defaults_put(deps: State, _: OwnerDep, body: GuestDefaultsBody):
    """The starting shortcuts a new guest gets and the skills every guest run
    is handed (sidecar ``shared_skills``) — a skill toggle in the catalog
    sends the whole list."""
    out = deps.users.set_defaults(body.model_dump(exclude_none=True))
    deps.hub.publish({"type": "users"})
    return out


@router.get("/api/guest-skills", responses=_r(schemas.GuestSkillsResponse))
def guest_skills(deps: State, caller: CallerDep):
    """The skills the caller's runs can use, for the composer's mentions: the
    owner's shared ones (as ``shared:<name>``, the plugin's namespace) and
    the granted projects' own ``.claude/skills``."""
    from dashboards import skills as skills_mod
    shared = deps.users.defaults()["sharedSkills"]
    cat = skills_mod.catalog()
    by_ref = {skills_mod.share_ref(s): s for s in cat}
    out, seen = [], set()
    for ref in shared:
        s = by_ref.get(ref)
        if s and s["name"] not in seen:
            seen.add(s["name"])
            out.append({"name": f"shared:{s['name']}", "description": s["description"],
                        "triggerWords": s["triggerWords"], "source": "shared"})
    for s in cat:
        if s["sourceKind"] == "project" and caller.can(s["source"], "chat"):
            out.append({"name": s["name"], "description": s["description"],
                        "triggerWords": s["triggerWords"], "source": s["source"]})
    return {"skills": out}


# ── conversation sharing ─────────────────────────────────────────────────────
def _share_state(deps, sid: str) -> dict:
    meta = deps.meta_store.peek(sid)
    slug = access.project_of(meta)
    owner = meta.get("owner")
    candidates = [{"id": u["id"], "name": u.get("name") or u["id"],
                   "level": (u.get("projects") or {}).get(slug)}
                  for u in deps.users.all()
                  if not u.get("disabled") and u["id"] != owner]
    return {"sessionId": sid, "owner": owner, "project": slug,
            "participants": list(meta.get("participants") or []),
            "candidates": candidates}


@router.get("/api/conversation-share", responses=_r(schemas.ConversationShare))
def share_get(deps: State, caller: CallerDep, id: str):
    sid = access.session_of(id)
    access.ensure_visible(deps, caller, sid)
    out = _share_state(deps, sid)
    if not caller.is_owner:
        # A guest only sees who's in it — not the owner's whole guest list.
        out["candidates"] = [c for c in out["candidates"] if c["level"]]
    return out


class ShareBody(BaseModel):
    id: str
    participants: list[str]


@router.put("/api/conversation-share", responses=_r(schemas.ConversationShare))
def share_put(deps: State, caller: CallerDep, body: ShareBody):
    """Who else may read and write this conversation. The owner can share
    anything with anyone; the guest who started a conversation can share it
    with users who already have access to its project."""
    sid = access.session_of(body.id)
    access.ensure_visible(deps, caller, sid)
    meta = deps.meta_store.peek(sid)
    slug = access.project_of(meta)
    if not caller.is_owner and meta.get("owner") != caller.id:
        raise HTTPException(403, "only the conversation's owner can share it")
    want = list(dict.fromkeys(p for p in body.participants if p))
    for uid in want:
        u = deps.users.active(uid)
        if not u:
            raise HTTPException(400, f"unknown user {uid!r}")
        if not caller.is_owner and not (u.get("projects") or {}).get(slug):
            raise HTTPException(403, f"{u.get('name') or uid} has no access to {slug!r}")
    deps.meta_store.update(sid, {"participants": [u for u in want
                                                  if u != meta.get("owner")]})
    deps.hub.publish({"type": "meta", "id": sid})
    return _share_state(deps, sid)