path: root/webmentions_ssg/views.py

# SPDX-FileCopyrightText: 2026 Dennis Fink <me+coding@dennisfink.me>
#
# SPDX-License-Identifier: BSD-3-Clause

import uuid
from typing import cast
from urllib.parse import urlsplit

import sqlalchemy as sa
import sqlalchemy.exc as sa_exc
from flask import (
    Blueprint,
    Response,
    abort,
    flash,
    redirect,
    render_template,
    request,
    url_for,
)
from flask.typing import ResponseReturnValue
from flask_login import current_user, login_required, login_user, logout_user
from sqlalchemy.orm import selectinload

from . import CSRF, forms
from . import DATABASE as db
from .models import ReceivedWebmention, Source, User
from .tasks.receiver import verify_webmention
from .tasks.scanner import manual_scan_sources

root_page = Blueprint("root", __name__)


@root_page.route("/")
def index() -> ResponseReturnValue:
    """
    Redirect to the received Webmentions view.

    :return: Redirect response to the received Webmentions view.
    """
    return redirect(url_for("root.received"))


@root_page.route("/login", methods=["GET", "POST"])
def login() -> ResponseReturnValue:
    """
    Authenticate a user and start a login session.

    Authenticated users are redirected to the application index. After a
    successful login, the user is redirected to the requested local URL when
    provided.

    :return: Rendered login page or redirect response.
    """
    if current_user.is_authenticated:
        return redirect(url_for("root.index"))

    form = forms.LoginForm()

    if form.validate_on_submit():
        user = db.session.scalar(
            sa.select(User).where(User.username == form.username.data)
        )

        if user is None or not user.check_password(form.password.data):
            flash("Invalid username or password", "danger")
            return redirect(url_for("root.login"))

        login_user(user)

        next_page = request.args.get("next")
        if not next_page or urlsplit(next_page).netloc != "":
            next_page = url_for("root.index")
        return redirect(next_page)

    return render_template("login.html", form=form)


@root_page.route("/logout")
def logout() -> ResponseReturnValue:
    """
    End the current user's login session.

    :return: Redirect response to the application index.
    """
    logout_user()
    return redirect(url_for("root.index"))


@root_page.route("/received")
@login_required
def received() -> ResponseReturnValue:
    """
    Display received Webmentions.

    :return: Rendered page containing the paginated received Webmentions.
    """
    webmentions = db.paginate(
        sa.select(ReceivedWebmention).order_by(ReceivedWebmention.uuid.desc()),
        per_page=25,
    )

    return render_template(
        "received.html",
        title="Received Webmentions",
        webmentions=webmentions,
        action_form=forms.AdminActionForm(),
    )


@root_page.post("/received/<uuid:identifier>/delete")
@login_required
def delete_received_webmention(identifier: uuid.UUID) -> ResponseReturnValue:
    """
    Delete a received Webmention.

    :param identifier: Identifier of the Webmention to delete.
    :return: Redirect response to the received Webmentions view.
    """
    form = forms.AdminActionForm()

    if not form.validate_on_submit():
        return abort(400)

    webmention = db.session.get(ReceivedWebmention, identifier)

    if webmention is None:
        return abort(404)

    db.session.delete(webmention)
    db.session.commit()

    flash(f"Webmention {webmention.uuid} deleted.", "success")

    return redirect(
        url_for("root.received", page=request.args.get("page", 1, type=int))
    )


@root_page.post("/received/<uuid:identifier>/reverify")
@login_required
def reverify_received_webmention(identifier: uuid.UUID) -> ResponseReturnValue:
    """
    Queue a received Webmention for reverification.

    The Webmention status is reset before a new verification task is queued.

    :param identifier: Identifier of the Webmention to reverify.
    :return: Redirect response to the received Webmentions view.
    """
    form = forms.AdminActionForm()

    if not form.validate_on_submit():
        return abort(400)

    webmention = db.session.get(ReceivedWebmention, identifier)

    if webmention is None:
        return abort(404)

    webmention.status = "received"
    webmention.failure_reason = None

    db.session.commit()

    verify_webmention(identifier)

    flash(f"Webmention {webmention.uuid} queued for reverification.", "success")

    return redirect(
        url_for("root.received", page=request.args.get("page", 1, type=int))
    )


@root_page.route("/sent")
@login_required
def sent() -> ResponseReturnValue:
    """
    Display sources with sent Webmentions.

    :return: Rendered page containing the paginated source list.
    """
    sources = db.paginate(
        sa.select(Source)
        .options(selectinload(Source.sent_webmentions))
        .order_by(Source.last_seen_at.desc()),
        per_page=25,
    )
    return render_template(
        "sent.html",
        title="Sent Webmentions",
        sources=sources,
        action_form=forms.AdminActionForm(),
    )


@root_page.post("/sent/rescan")
@login_required
def rescan_sent_sources() -> ResponseReturnValue:
    """
    Queue a manual scan of sent Webmention sources.

    :return: Redirect response to the sent Webmentions view.
    """
    form = forms.AdminActionForm()
    if form.validate_on_submit():
        manual_scan_sources()
        flash("Source rescan queued.", "success")
    return redirect(url_for("root.sent"))


@root_page.route("/sent/<uuid:identifier>")
@login_required
def sent_source(identifier: uuid.UUID) -> ResponseReturnValue:
    """
    Display the sent Webmentions associated with a source.

    :param identifier: Identifier of the source to display.
    :return: Rendered source details page.
    """
    source = db.session.scalar(
        sa.select(Source)
        .options(selectinload(Source.sent_webmentions))
        .where(Source.uuid == identifier)
    )
    if source is None:
        return abort(404)

    webmentions = sorted(
        source.sent_webmentions, key=lambda webmention: webmention.target
    )
    return render_template(
        "sent_source.html",
        title="Sent Webmentions",
        source=source,
        webmentions=webmentions,
    )


@root_page.post("/endpoint")
@CSRF.exempt
def endpoint() -> ResponseReturnValue:
    """
    Receive and queue a Webmention for verification.

    Existing Webmentions with the same source and target are reset for
    reverification. Concurrent insertion of the same Webmention is handled by
    retrieving and updating the row created by the competing request.

    :return: HTTP 201 response with the Webmention status URL in the
        ``Location`` header, or validation errors with HTTP 400.
    """
    form = forms.EndpointForm(meta={"csrf": False})

    if not form.validate_on_submit():
        return form.errors, 400

    source = cast(str, form.source.data)
    target = cast(str, form.target.data)

    webmention = db.session.execute(
        sa.select(ReceivedWebmention).where(
            ReceivedWebmention.source == source, ReceivedWebmention.target == target
        )
    ).scalar_one_or_none()

    if webmention is None:
        identifier = uuid.uuid7()

        webmention = ReceivedWebmention(uuid=identifier, source=source, target=target)

        db.session.add(webmention)

        try:
            db.session.commit()
        except sa_exc.IntegrityError:
            db.session.rollback()

            # Another request may have inserted the same source/target pair
            # after our SELECT but before our COMMIT.
            webmention = db.session.execute(
                sa.select(ReceivedWebmention).where(
                    ReceivedWebmention.source == source,
                    ReceivedWebmention.target == target,
                )
            ).scalar_one()

            identifier = webmention.uuid
            webmention.status = "received"
            webmention.failure_reason = None

            db.session.commit()

    else:
        identifier = webmention.uuid

        # Re-sent webmention: re-verify the existing row.
        webmention.status = "received"
        webmention.failure_reason = None

        db.session.commit()

    verify_webmention(webmention.uuid)

    status_url = url_for("root.status", identifier=str(identifier), _external=True)

    return Response(status=201, headers={"Location": status_url})


@root_page.route("/status/<uuid:identifier>")
def status(identifier: uuid.UUID) -> ResponseReturnValue:
    """
    Display the verification status of a received Webmention.

    :param identifier: Identifier of the Webmention to display.
    :return: Rendered Webmention status page.
    """
    webmention = db.session.get(ReceivedWebmention, identifier)

    if webmention is None:
        return abort(404)

    return render_template(
        "status.html", title="Webmention status", webmention=webmention
    )