federation/bp/auth/routes.py

"""Authentication routes for the federation app.

Ported from blog/bp/auth/routes.py — owns magic link login/logout.
Simplified: no Ghost sync, no newsletter management (those stay in blog).
"""
from __future__ import annotations

from datetime import datetime, timezone

from quart import (
    Blueprint,
    request,
    render_template,
    make_response,
    redirect,
    url_for,
    session as qsession,
    g,
    current_app,
)
from sqlalchemy import select
from sqlalchemy.exc import SQLAlchemyError

from shared.db.session import get_session
from shared.models import User
from shared.config import config
from shared.utils import host_url
from shared.infrastructure.urls import federation_url
from shared.infrastructure.cart_identity import current_cart_identity
from shared.events import emit_activity
from shared.services.registry import services

from .services import (
    pop_login_redirect_target,
    store_login_redirect_target,
    send_magic_email,
    find_or_create_user,
    create_magic_link,
    validate_magic_link,
    validate_email,
)

SESSION_USER_KEY = "uid"


def register(url_prefix="/auth"):
    auth_bp = Blueprint("auth", __name__, url_prefix=url_prefix)

    @auth_bp.get("/login/")
    async def login_form():
        store_login_redirect_target()
        cross_cart_sid = request.args.get("cart_sid")
        if cross_cart_sid:
            qsession["cart_sid"] = cross_cart_sid
        if g.get("user"):
            return redirect(federation_url("/"))
        return await render_template("auth/login.html")

    @auth_bp.get("/account/")
    async def account():
        if not g.get("user"):
            return redirect(host_url(url_for("auth.login_form")))

        # Check if user has an ActorProfile
        actor = await services.federation.get_actor_by_user_id(g.s, g.user.id)
        return await render_template(
            "federation/account.html",
            actor=actor,
        )

    @auth_bp.post("/start/")
    async def start_login():
        form = await request.form
        email_input = form.get("email") or ""

        is_valid, email = validate_email(email_input)
        if not is_valid:
            return (
                await render_template(
                    "auth/login.html",
                    error="Please enter a valid email address.",
                    email=email_input,
                ),
                400,
            )

        user = await find_or_create_user(g.s, email)
        token, expires = await create_magic_link(g.s, user.id)

        magic_url = host_url(url_for("auth.magic", token=token))

        email_error = None
        try:
            await send_magic_email(email, magic_url)
        except Exception as e:
            current_app.logger.error("EMAIL SEND FAILED: %r", e)
            email_error = (
                "We couldn't send the email automatically. "
                "Please try again in a moment."
            )

        return await render_template(
            "auth/check_email.html",
            email=email,
            email_error=email_error,
        )

    @auth_bp.get("/magic/<token>/")
    async def magic(token: str):
        now = datetime.now(timezone.utc)
        user_id: int | None = None

        try:
            async with get_session() as s:
                async with s.begin():
                    user, error = await validate_magic_link(s, token)

                    if error:
                        return (
                            await render_template("auth/login.html", error=error),
                            400,
                        )
                    user_id = user.id

        except Exception:
            return (
                await render_template(
                    "auth/login.html",
                    error="Could not sign you in right now. Please try again.",
                ),
                502,
            )

        assert user_id is not None

        ident = current_cart_identity()
        anon_session_id = ident.get("session_id")

        try:
            async with get_session() as s:
                async with s.begin():
                    u2 = await s.get(User, user_id)
                    if u2:
                        u2.last_login_at = now
                    if anon_session_id:
                        await emit_activity(
                            s,
                            activity_type="rose:Login",
                            actor_uri="internal:system",
                            object_type="Person",
                            object_data={
                                "user_id": user_id,
                                "session_id": anon_session_id,
                            },
                        )
        except SQLAlchemyError:
            current_app.logger.exception(
                "[auth] non-fatal DB update for user_id=%s", user_id
            )

        qsession[SESSION_USER_KEY] = user_id

        redirect_url = pop_login_redirect_target()
        return redirect(redirect_url, 303)

    @auth_bp.post("/logout/")
    async def logout():
        qsession.pop(SESSION_USER_KEY, None)
        return redirect(federation_url("/"))

    return auth_bp