mqtt-broker-stack/backend/app/routers/topics.py

"""Topic explorer API (spec sections 35, 36, 41).

Endpoints:
    POST /topics/subscribe       -> open an explorer session
    GET  /topics/stream/{id}      -> Server-Sent Events message stream
    POST /topics/publish          -> publish a test message (audited)
    GET  /topics/retained         -> list retained messages (spec 36)
    POST /topics/retained/clear   -> clear one retained message (audited)

Security model (spec section 35: access restricted through management
RBAC; all publishes audited):
    * subscribe/stream/retained listing: OPERATOR and above (spec
      section 9: "perform permitted diagnostic operations")
    * publish / clear retained: BROKER_ADMIN and above (state-changing)
    * every publish and clear writes an audit event with topic and QoS
      metadata — NEVER the payload (audit rows are not a message store).
"""

from __future__ import annotations

import asyncio
import base64
import json
from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, Request, status
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session

from app import audit, explorer
from app.db import get_db
from app.deps import CurrentUser, require_roles
from app.models import ManagementRole

router = APIRouter(prefix="/api/v1/topics", tags=["topics"])

OperatorRole = Annotated[
    CurrentUser,
    Depends(
        require_roles(
            ManagementRole.BROKER_ADMIN,
            ManagementRole.OPERATOR,
        )
    ),
]
BrokerAdmin = Annotated[CurrentUser, Depends(require_roles(ManagementRole.BROKER_ADMIN))]
DbDep = Annotated[Session, Depends(get_db)]


class SubscribeRequest(BaseModel):
    topic_filter: str = Field(min_length=1, max_length=200)


class SubscribeResponse(BaseModel):
    session_id: str


class PublishRequest(BaseModel):
    topic: str = Field(min_length=1, max_length=200)
    payload_b64: str = Field(default="", description="Empty clears retained")
    qos: int = Field(default=0, ge=0, le=2)
    retain: bool = False


class RetainedMessage(BaseModel):
    index: int
    topic: str
    qos: int
    retain: bool
    received_at: float
    payload_b64: str
    payload_text: str | None
    properties: dict | None = None


class ClearRetainedRequest(BaseModel):
    topic: str = Field(min_length=1, max_length=200)


def _valid_topic_filter(value: str) -> None:
    # Reject wildcard misuse the broker would reject anyway, with a
    # clearer error than a generic MQTT rc.
    if "#" in value and not value.endswith("#"):
        raise HTTPException(
            status.HTTP_400_BAD_REQUEST,
            "'#' is only valid as the last level of a filter",
        )


@router.post("/subscribe", response_model=SubscribeResponse)
def subscribe(body: SubscribeRequest, user: OperatorRole):
    """Open an explorer session for one topic filter (spec section 35)."""
    _valid_topic_filter(body.topic_filter)
    explorer.sessions.sweep()
    try:
        session = explorer.sessions.start(user.username, body.topic_filter)
    except OSError as exc:
        raise HTTPException(
            status.HTTP_503_SERVICE_UNAVAILABLE,
            f"could not connect to the broker: {exc}",
        ) from exc
    return SubscribeResponse(session_id=session.id)


@router.get("/stream/{session_id}")
async def stream(session_id: str, user: OperatorRole):
    """Server-Sent Events stream of live messages for the session.

    Closing the EventSource stops the session (and its MQTT client) via
    the generator's finally — the lifecycle is tied to the connection.
    """
    try:
        explorer.sessions.get(session_id)
    except KeyError as exc:
        raise HTTPException(status.HTTP_404_NOT_FOUND, "session not found") from exc

    async def event_stream():
        # Start BELOW the first index: messages_since uses strictly-
        # greater semantics, and a fresh session's first message has
        # index 0 — a cursor starting at 0 would drop it.
        index = -1
        heartbeat = 0.0
        try:
            while True:
                for message in explorer.messages_since(session_id, index):
                    index = message["index"]
                    yield f"data: {json.dumps(message)}\n\n"
                # Heartbeat comment keeps proxies from idling out.
                heartbeat += 1
                if heartbeat % 60 == 0:  # ~15s at 0.25s cadence
                    yield ": keepalive\n\n"
                await asyncio.sleep(0.25)
        finally:
            explorer.sessions.stop(session_id)

    return StreamingResponse(
        event_stream(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",  # nginx: disable response buffering
        },
    )


@router.post("/publish", status_code=status.HTTP_204_NO_CONTENT)
def publish(
    body: PublishRequest,
    request: Request,
    db: DbDep,
    user: BrokerAdmin,
):
    """Publish a test message (spec section 35: all publishes audited).

    An empty payload with retain=true is MQTT's clear-retained mechanism
    (spec section 36) — the same call serves both purposes.
    """
    _valid_topic_filter(body.topic)
    if any(c in body.topic for c in ("#", "+")):
        raise HTTPException(
            status.HTTP_400_BAD_REQUEST,
            "wildcards cannot be published to — use a concrete topic",
        )
    try:
        payload = base64.b64decode(body.payload_b64) if body.payload_b64 else b""
    except Exception as exc:
        raise HTTPException(status.HTTP_400_BAD_REQUEST, "invalid base64 payload") from exc
    try:
        explorer.publish_message(body.topic, payload, body.qos, body.retain)
    except (OSError, RuntimeError) as exc:
        raise HTTPException(
            status.HTTP_503_SERVICE_UNAVAILABLE,
            f"publish failed: {exc}",
        ) from exc
    audit.record(
        db,
        "TOPIC_PUBLISHED",
        request=request,
        user=user.username,
        object_type="topic",
        object_id=body.topic,
        new_value={"qos": body.qos, "retain": body.retain, "bytes": len(payload)},
    )
    return None


@router.get("/retained", response_model=list[RetainedMessage])
def retained(user: OperatorRole):
    """List retained messages (spec section 36).

    Brokers have no list-retained API: this subscribes briefly to #,
    collects retain-flag deliveries, and unsubscribes. Capped; large
    brokers deliver everything on subscribe.
    """
    try:
        return explorer.list_retained()
    except OSError as exc:
        raise HTTPException(
            status.HTTP_503_SERVICE_UNAVAILABLE,
            f"could not connect to the broker: {exc}",
        ) from exc


@router.post("/retained/clear", status_code=status.HTTP_204_NO_CONTENT)
def clear_retained(
    body: ClearRetainedRequest,
    request: Request,
    db: DbDep,
    user: BrokerAdmin,
):
    """Clear one retained message: publish a zero-length payload with the
    retain flag (MQTT's defined mechanism, spec section 36)."""
    try:
        explorer.publish_message(body.topic, b"", qos=0, retain=True)
    except (OSError, RuntimeError) as exc:
        raise HTTPException(
            status.HTTP_503_SERVICE_UNAVAILABLE,
            f"clear failed: {exc}",
        ) from exc
    audit.record(
        db,
        "RETAINED_CLEARED",
        request=request,
        user=user.username,
        object_type="topic",
        object_id=body.topic,
    )
    return None