feat(pelican): stream file events over SSE at GET /pelican/subscribe

CHANGELOG.md

@@ -7,6 +7,25 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- **`GET /pelican/subscribe` streams Pelican file events as Server-Sent Events.** The event server notifies subscribers whenever an object appears in a Pelican namespace, but that stream existed only in the `ndp-ep` client library; the API had no way to expose it, so an Endpoint could not offer subscriptions to its own callers. The new route returns a `text/event-stream` where each notification arrives as an `event: file` message carrying the object's `name`, `url`, `size` and `mod_time`. The `url` goes straight to `/pelican/read` for the contents or `/pelican/download` for the file, which is the pipeline the route exists to enable: subscribe, then read what arrived. A `: keepalive` comment is emitted during quiet periods so a proxy does not mistake an idle stream for a dead one, and `X-Accel-Buffering: no` stops nginx holding events back until its buffer fills.
+- **`GET /pelican/subscriptions` reports the upstream subscriptions the Endpoint holds**, with each one's connection state, event source, client id, listener count and how many events were dropped for slow listeners. No credential appears in it — the username is reported only as an `authenticated` flag — since any viewer on the Endpoint can read the route.
+- **A caller may bring its own event server identity and credentials.** `client_id`, `username` and `password` are accepted as query parameters, and as the `X-Pelican-Event-Client-Id`, `X-Pelican-Event-Username` and `X-Pelican-Event-Password` headers, which take precedence; anything omitted falls back to the Endpoint's own configuration, so callers with an account of their own and callers without one are served by the same route. The headers exist because a query string is written to the access logs of both uvicorn and nginx, which would put a password on disk in plain text. Credentials are taken as a pair: supplying only a username is refused rather than silently borrowing the Endpoint's password, which would sign the caller in as the Endpoint under a name of their choosing.
+
+### Changed
+- The event server speaks STOMP 1.2 over a WebSocket, so `websockets` is now a dependency.
+
+### Notes on the design
+- **One upstream subscription per destination, fanned out to every listener.** The event server requires a unique `client-id` per subscriber: two connections sharing one are served by splitting the events between them, so each sees only a fraction. Opening a subscription per SSE caller would therefore either split the stream that way or leave an orphaned client id registered upstream on every connection. Instead the upstream is keyed by client id *and* event source: subscribers presenting the same identity share one connection and each receive every event on it, while a caller bringing its own credentials gets its own, since two identities cannot be served over a single authenticated session. The connection opens when the first listener arrives and closes when the last one leaves, so an idle Endpoint holds none at all.
+- **The STOMP protocol is implemented here rather than taken from the client library.** Depending on `ndp-ep` would make the API depend on its own client SDK and pull in `pelicanfs`, which pins Python 3.11. The cost is that the wire format now lives in two repositories and has to be kept in step by hand.
+- **Per-listener buffers are bounded**, unlike the client library's. A caller that stops reading must not be able to grow the Endpoint's memory without limit, so a full buffer drops its oldest event and counts it in `/pelican/subscriptions`.
+
+### Known limitation
+- **The broker is per worker process, not per Endpoint.** `Dockerfile.allinone` runs uvicorn with four workers, and each holds its own broker. Two callers relying on the Endpoint's configured client id can therefore land on different workers and open two upstream connections under one identity, which the event server serves by splitting the events between them — so each would see only part of the stream. Callers that supply their own client id are unaffected, and so is any deployment running a single worker. Fixing it properly needs the subscription to live outside the worker processes.
+
+### Backwards compatibility
+- Purely additive. Both routes sit behind the authorization added in #261 and are only mounted when `PELICAN_ENABLED` is set. Subscriptions need a client id from somewhere — the request, `PELICAN_EVENT_CLIENT_ID`, or an `AFFINITIES_EP_UUID` to derive it from — and `/pelican/subscribe` answers 503 with the reason when none is available, so an Endpoint that never configures the event server is unaffected. The other new settings (`PELICAN_EVENT_SERVER_URL`, `PELICAN_EVENT_USERNAME`, `PELICAN_EVENT_PASSWORD`, `PELICAN_EVENT_VIRTUAL_HOST`, `PELICAN_EVENT_HEARTBEAT_MS`) are optional and documented in `example.env` and `docs/configuration.md`.
+
 ## [0.34.23] - 2026-08-30
 
 ### Added

api/routes/pelican_routes.py

@@ -5,7 +5,15 @@
 These endpoints allow browsing and downloading from external Pelican federations.
 """
 
-from fastapi import APIRouter, Depends, HTTPException, Query, Response
+from fastapi import (
+    APIRouter,
+    Depends,
+    Header,
+    HTTPException,
+    Query,
+    Request,
+    Response,
+)
 from fastapi.responses import StreamingResponse
 from pydantic import BaseModel
 from typing import Optional, Dict, Any
@@ -16,11 +24,17 @@
 )
 from api.services.pelican_services.download_file import download_file, stream_file
 from api.services.pelican_services.read_file import read_object
+from api.services.pelican_services.event_subscription import (
+    EventSubscriptionUnavailable,
+    broker,
+    load_config,
+)
 from api.services.pelican_services.import_metadata import import_file_as_resource
 from api.services.auth_services import (
     get_user_for_read_operation,
     get_user_for_write_operation,
 )
+import json
 import logging
 import os
 
@@ -339,6 +353,141 @@ async def read_file_contents(
         raise HTTPException(status_code=500, detail=f"Error reading object: {str(e)}")
 
 
+def _sse(event: str, payload: dict) -> str:
+    """Format one Server-Sent Event message."""
+    return f"event: {event}\ndata: {json.dumps(payload)}\n\n"
+
+
+@router.get("/subscribe")
+async def subscribe_to_events(
+    request: Request,
+    event_source: str = Query(
+        ..., description="Namespace to watch, e.g. osdf/vdc/public/data"
+    ),
+    client_id: Optional[str] = Query(
+        None,
+        description=(
+            "Identity to present to the event server. Must be unique: "
+            "two subscribers sharing one are served by splitting the "
+            "events between them. Defaults to the Endpoint's own."
+        ),
+    ),
+    username: Optional[str] = Query(
+        None,
+        description=(
+            "Event server username. Prefer the X-Pelican-Event-Username "
+            "header. Defaults to the Endpoint's own credentials."
+        ),
+    ),
+    password: Optional[str] = Query(
+        None,
+        description=(
+            "Event server password. Prefer the X-Pelican-Event-Password "
+            "header: a query string is written to the access log."
+        ),
+    ),
+    header_client_id: Optional[str] = Header(None, alias="X-Pelican-Event-Client-Id"),
+    header_username: Optional[str] = Header(None, alias="X-Pelican-Event-Username"),
+    header_password: Optional[str] = Header(None, alias="X-Pelican-Event-Password"),
+):
+    """
+    Stream Pelican file events as Server-Sent Events.
+
+    Each event arrives as an ``event: file`` message whose data carries
+    the object's ``name``, ``url``, ``size`` and ``mod_time``. The
+    ``url`` can be handed straight to ``/pelican/read`` to get the
+    contents, or to ``/pelican/download`` to fetch it as a file.
+
+    A ``: keepalive`` comment is sent during quiet periods, so a proxy
+    does not mistake an idle stream for a dead one.
+
+    The caller may bring its own event server identity and credentials;
+    anything it omits falls back to the Endpoint's configuration. They
+    are accepted both as query parameters and as headers, and the
+    headers win. **Prefer the headers**: a query string is recorded in
+    the access log of both uvicorn and nginx, so a password passed that
+    way is written to disk in plain text.
+
+    Subscribers presenting the same client id share one upstream
+    connection and each receive every event on it. A caller with its own
+    id gets its own connection, since the two cannot be served over a
+    single authenticated session.
+
+    Parameters
+    ----------
+    request : Request
+        Used to notice that the caller has gone away.
+    event_source : str
+        Namespace to watch.
+    client_id, username, password : str, optional
+        Event server identity and credentials, overriding the
+        Endpoint's.
+    header_client_id, header_username, header_password : str, optional
+        The same three, taken from headers, which take precedence.
+
+    Returns
+    -------
+    StreamingResponse
+        A ``text/event-stream``.
+
+    Raises
+    ------
+    HTTPException
+        503 if no usable event server configuration results.
+    """
+    try:
+        config = load_config(
+            client_id=header_client_id or client_id,
+            username=header_username or username,
+            password=header_password or password,
+        )
+    except EventSubscriptionUnavailable as exc:
+        raise HTTPException(status_code=503, detail=str(exc))
+
+    async def event_stream():
+        # Sent immediately, so the caller can tell the stream is open
+        # even while the namespace is quiet.
+        yield ": subscribed\n\n"
+        try:
+            async for event in broker.listen(event_source, config):
+                if await request.is_disconnected():
+                    break
+                if event is None:
+                    yield ": keepalive\n\n"
+                    continue
+                yield _sse("file", event)
+        except EventSubscriptionUnavailable as exc:
+            yield _sse("error", {"detail": str(exc)})
+        except Exception as exc:  # pragma: no cover - defensive
+            logger.error(f"Pelican event stream failed: {exc}")
+            yield _sse("error", {"detail": f"{type(exc).__name__}: {exc}"})
+
+    return StreamingResponse(
+        event_stream(),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            # Without this nginx buffers the stream and holds events
+            # back until the buffer fills, defeating the point.
+            "X-Accel-Buffering": "no",
+        },
+    )
+
+
+@router.get("/subscriptions")
+async def list_subscriptions():
+    """
+    Report the upstream subscriptions this Endpoint currently holds.
+
+    Returns
+    -------
+    dict
+        One entry per event source with its connection state, listener
+        count, and how many events were dropped for slow listeners.
+    """
+    return {"success": True, "subscriptions": broker.status()}
+
+
 @router.post("/import-metadata")
 async def import_metadata(
     request: ImportMetadataRequest,