path: root/jb/views/auth.py

import secrets
from typing import Annotated
from urllib.parse import urlencode

from fastapi import (
    APIRouter,
    BackgroundTasks,
    Depends,
    Header,
    HTTPException,
    Request,
    Response,
    status,
)
from fastapi.responses import HTMLResponse, RedirectResponse

from jb.api.auth import (
    SESSION_COOKIE_NAME,
    create_session,
    get_authenticated_user,
)
from jb.api.magic_token import (
    INVITE_AMT_LINK_TOKEN_TTL,
    consume_amt_account_link_token,
    consume_magic_token,
    create_amt_account_link_token,
    create_magic_token,
)
from jb.config import settings
from jb.decorators import LOG
from jb.dependencies import get_gr_api_manager
from jb.managers.email_manager import (
    get_or_create_contact,
    send_amt_link_email,
    send_login_email,
)
from jb.managers.gr_api import GRApiManager
from jb.managers.thl import create_paypal_cashout_method_if_not_exists
from jb.models.auth import (
    AccountLogin,
    AmtAccountLink,
    MagicLinkExchangeRequest,
    User,
)
from jb.settings import render_base_html
from jb.views.utils import get_client_ip

auth_router = APIRouter(prefix="/auth", tags=["Auth"])


def try_create_paypal_cashout_method_if_not_exists(
    product_user_id: str, email: str, client_ip: str
) -> None:
    try:
        create_paypal_cashout_method_if_not_exists(
            product_user_id=product_user_id,
            email=email,
            client_ip=client_ip,
        )
    except Exception:
        LOG.exception("Failed to create PayPal cashout method for %s", product_user_id)


def authenticate_invite_amt_account_link(
    authorization: Annotated[str | None, Header()] = None,
) -> None:
    expected_token = settings.invite_amt_account_token
    if expected_token is None:
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail="not configured",
        )

    scheme, _, supplied_token = (authorization or "").partition(" ")
    authenticated = scheme.lower() == "bearer" and secrets.compare_digest(
        supplied_token,
        expected_token.get_secret_value(),
    )
    if not authenticated:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid scheduler authentication token",
            headers={"WWW-Authenticate": "Bearer"},
        )


@auth_router.post("/magic-link/request/")
def request_magic_link(body: AccountLogin) -> dict[str, str]:
    return _request_magic_link(str(body.email))


def _request_magic_link(email: str) -> dict[str, str]:
    token = create_magic_token(user_email=email)

    if settings.debug:
        query = urlencode({"token": token})
        return {"magic_link": f"{settings.base_url}auth/magic-link/?{query}"}

    send_login_email(email=email, magic_token=token)
    return {"detail": "Link sent. Check your inbox and follow the link to log in."}


@auth_router.get("/magic-link/", response_class=HTMLResponse, include_in_schema=False)
def magic_link_landing_page(
    request: Request,
    background_tasks: BackgroundTasks,
    gr_api: Annotated[GRApiManager, Depends(get_gr_api_manager)],
    token: str | None = None,
) -> Response:
    """Serve the SPA without redeeming the token; email prefetches are harmless."""
    if settings.debug:
        if token is None:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="token is required",
            )
        response = RedirectResponse(url="/", status_code=status.HTTP_303_SEE_OTHER)
        client_ip = get_client_ip(request)
        _exchange_magic_link(
            token,
            response,
            gr_api,
            client_ip=client_ip,
            background_tasks=background_tasks,
        )
        return response
    return HTMLResponse(
        render_base_html(),
        headers={
            "Cache-Control": "no-store",
            "Referrer-Policy": "no-referrer",
            "X-Robots-Tag": "noindex, nofollow",
        },
    )


@auth_router.post("/magic-link/exchange/", status_code=status.HTTP_204_NO_CONTENT)
def exchange_magic_link(
    request: Request,
    background_tasks: BackgroundTasks,
    body: MagicLinkExchangeRequest,
    response: Response,
    gr_api: Annotated[GRApiManager, Depends(get_gr_api_manager)],
) -> None:
    """Exchange a magic link only after its landing page makes an explicit POST."""
    client_ip = get_client_ip(request)
    _exchange_magic_link(
        body.token,
        response,
        gr_api,
        client_ip=client_ip,
        background_tasks=background_tasks,
    )


def _exchange_magic_link(
    token: str,
    response: Response,
    gr_api: GRApiManager,
    client_ip: str,
    background_tasks: BackgroundTasks,
) -> None:
    user_email = consume_magic_token(token)
    user = gr_api.ensure_user_exists(User.model_validate({"email": user_email}))

    # Cashout setup is not required for login and should not delay the response.
    background_tasks.add_task(
        try_create_paypal_cashout_method_if_not_exists,
        product_user_id=user.product_user_id,
        email=str(user.email),
        client_ip=client_ip,
    )

    response.set_cookie(
        key=SESSION_COOKIE_NAME,
        value=create_session(user.product_user_id),
        max_age=settings.session_token_ttl_seconds,
        httponly=True,
        secure=not settings.debug,
        samesite="lax",
        path="/",
    )


@auth_router.post("/link-amt/request/")
def link_amt_account(
    body: AmtAccountLink,
    gr_api: Annotated[GRApiManager, Depends(get_gr_api_manager)],
) -> dict[str, str]:
    """Link an AMT account and login."""
    email = str(body.email)
    amt_worker_id = body.amt_worker_id

    # If the email already has transitioned, treat this as a login request
    user = User.model_validate({"email": email})
    user_exists = gr_api.get_user_if_exists(user.product_user_id)
    if user_exists:
        return _request_magic_link(email)

    token = create_amt_account_link_token(email=email, amt_worker_id=amt_worker_id)

    if settings.debug:
        query = urlencode({"token": token})
        return {"magic_link": f"{settings.base_url}auth/link-amt/?{query}"}

    send_amt_link_email(email=email, magic_token=token)
    return {}


@auth_router.post("/link-amt/invite/")
def invite_amt_account_link(
    body: AmtAccountLink,
    _authenticated: Annotated[None, Depends(authenticate_invite_amt_account_link)],
) -> dict[str, str]:
    """Create a long-lived AMT account-link URL without sending an email."""
    token = create_amt_account_link_token(
        email=str(body.email),
        amt_worker_id=body.amt_worker_id,
        ttl_seconds=INVITE_AMT_LINK_TOKEN_TTL,
    )
    query = urlencode({"token": token})
    return {"magic_link": f"{settings.base_url}auth/link-amt/?{query}"}


@auth_router.get("/debug/", response_class=HTMLResponse, include_in_schema=False)
def link_amt_account_landing_page(
    request: Request,
    background_tasks: BackgroundTasks,
    gr_api: Annotated[GRApiManager, Depends(get_gr_api_manager)],
    token: str | None = None,
) -> HTMLResponse:
    """Serve the account-link SPA without consuming the one-time token."""

    # TODO! Try catch any of this, and if it fails, show the user a
    # TODO! failed HTML page. As of now, it shows them a failed JSON response.

    if settings.debug:
        if token is None:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="token is required",
            )
        client_ip = get_client_ip(request)
        _response = RedirectResponse(url="/", status_code=status.HTTP_303_SEE_OTHER)
        _exchange_amt_account_link(
            token=token,
            response=_response,
            gr_api=gr_api,
            client_ip=client_ip,
            background_tasks=background_tasks,
        )

    return HTMLResponse(
        render_base_html(),
        headers={
            "Cache-Control": "no-store",
            "Referrer-Policy": "no-referrer",
            "X-Robots-Tag": "noindex, nofollow",
        },
    )


@auth_router.post("/link-amt/exchange/", status_code=status.HTTP_204_NO_CONTENT)
def exchange_amt_account_link(
    request: Request,
    background_tasks: BackgroundTasks,
    body: MagicLinkExchangeRequest,
    response: Response,
    gr_api: Annotated[GRApiManager, Depends(get_gr_api_manager)],
) -> None:
    """Validate the email link, then transition the bound AMT account."""
    client_ip = get_client_ip(request)
    try:
        _exchange_amt_account_link(
            body.token,
            response,
            gr_api,
            client_ip=client_ip,
            background_tasks=background_tasks,
        )
    except ValueError as e:
        LOG.error(f"Failed to exchange AMT account link: {e}")
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))


def _exchange_amt_account_link(
    token: str,
    response: Response,
    gr_api: GRApiManager,
    client_ip: str,
    background_tasks: BackgroundTasks,
):
    token_data = consume_amt_account_link_token(token)
    email = token_data.email
    amt_worker_id = token_data.amt_worker_id

    user = User(email=email)
    user = gr_api.transition_user_from_amt(user=user, amt_worker_id=amt_worker_id)
    background_tasks.add_task(
        try_create_paypal_cashout_method_if_not_exists,
        product_user_id=user.product_user_id,
        email=str(user.email),
        client_ip=client_ip,
    )

    # In Mautic, associate the email with the worker ID (AFTER the user has transitioned)
    get_or_create_contact(email=email, amt_worker_id=amt_worker_id)

    session_token = create_session(user.product_user_id)
    response.set_cookie(
        key=SESSION_COOKIE_NAME,
        value=session_token,
        max_age=settings.session_token_ttl_seconds,
        httponly=True,
        secure=not settings.debug,
        samesite="lax",
        path="/",
    )


@auth_router.get("/session/", response_model=User)
def get_session(
    user: Annotated[User, Depends(get_authenticated_user)],
) -> User:
    return user


@auth_router.delete("/session/", status_code=status.HTTP_204_NO_CONTENT)
def delete_session(
    response: Response,
) -> None:
    # Logout is idempotent so clients can always discard their local token.
    # JWT sessions are stateless, so logout discards the browser cookie. A token
    # copied elsewhere remains valid until its short, configured expiration.
    response.delete_cookie(key=SESSION_COOKIE_NAME, path="/")