bdlg2023-sms-back/app/api/endpoints/project_stream.py

import json

from fastapi import APIRouter, Depends, Request
from fastapi.responses import StreamingResponse

from app.api import deps
from app.core.events import broadcast
from app.models import OrgRole, User

router = APIRouter()


@router.get("/project/{project_id}/stream", summary="Real-Time SSE Stream")
async def project_stream(
    project_id: str,
    client_id: str,
    request: Request,
    current_user: User = Depends(
        deps.require_org_role(OrgRole.ORG_ADMIN, OrgRole.RESPO_BENEVOLE, OrgRole.RESPO_COMMISSION)
    ),
):
    """
    Connect to this endpoint to receive real-time updates.

    **Required Query Parameters:**
    - `client_id`: A unique UUID generated by the frontend to prevent echoing.
    """

    async def event_generator():
        channel = f"project_{project_id}"

        # Subscribe to the Redis channel specific to this project
        async with broadcast.subscribe(channel=channel) as subscriber:
            async for event in subscriber:
                # If the user closes the tab or navigates away, cleanly close the connection
                if await request.is_disconnected():
                    break
                # Parse the raw string payload from Redis
                message = json.loads(event.message)

                # Echo prevention: Do not send the event back to the client that caused it
                if message.get("client_id") == client_id:
                    continue

                # Yield the properly formatted Server-Sent Event (SSE)
                yield f"event: {message['event']}\ndata: {json.dumps(message['item'])}\n\n"

    # Return as text/event-stream so the browser knows it's an ongoing connection
    return StreamingResponse(event_generator(), media_type="text/event-stream")