diff --git a/README.md b/README.md index 9ed45db..1f75536 100644 --- a/README.md +++ b/README.md @@ -151,6 +151,42 @@ sso = FastCommentsSSO.new_simple(user_data) sso_token = sso.create_token() ``` +### Live Subscriptions (PubSub) + +The `pubsub` module lets you subscribe to real-time comment events (new comments, votes, edits, notifications, etc.) over a WebSocket, mirroring the FastComments Java SDK's `LiveEventSubscriber`. It requires the `pubsub` extra, which adds a WebSocket client on top of the generated API client: + +```bash +pip install "fastcomments[pubsub] @ git+https://github.com/fastcomments/fastcomments-python.git@v3.0.0" +``` + +```python +from pubsub import LiveEventSubscriber + +subscriber = LiveEventSubscriber() + +def handle_live_event(event): + print(f"Live event: {event.type}") + if event.comment: + print(f" comment: {event.comment.comment}") + +result = subscriber.subscribe_to_changes( + tenant_id_ws="YOUR_TENANT_ID", + url_id="page-url-id", + url_id_ws="page-url-id", + user_id_ws="a-unique-presence-id", # e.g. a UUID for this session + handle_live_event=handle_live_event, + on_connection_status_change=lambda connected, last_event_time: print( + f"connected={connected}" + ), + region=None, # set to "eu" for the EU region +) + +# ...later, when you no longer want updates: +result.close() +``` + +The subscriber runs the connection on a background daemon thread, transparently reconnects with jitter, and fetches any events missed while disconnected from the event-log endpoint on reconnect. Pass an optional `can_see_comments` callback (`List[str] -> Dict[str, str]`, returning the ids the user may NOT see) to filter out events for comments the user is not allowed to view. Set `disable_live_commenting=True` to make `subscribe_to_changes` a no-op that returns `None`. + ### Common Issues 1. **401 "missing-api-key" error**: Make sure you set `config.api_key = {"api_key": "YOUR_KEY"}` before creating the DefaultApi instance. @@ -158,6 +194,7 @@ sso_token = sso.create_token() 3. **Import errors**: Make sure you're importing from the correct module: - API client: `from client import ...` - SSO utilities: `from sso import ...` + - Live subscriptions: `from pubsub import ...` (needs the `pubsub` extra) ## Development diff --git a/pubsub/__init__.py b/pubsub/__init__.py new file mode 100644 index 0000000..fe1e131 --- /dev/null +++ b/pubsub/__init__.py @@ -0,0 +1,9 @@ +"""FastComments live pub/sub: subscribe to real-time comment events over WebSocket.""" + +from .live_event_subscriber import LiveEventSubscriber +from .subscribe_to_changes_result import SubscribeToChangesResult + +__all__ = [ + "LiveEventSubscriber", + "SubscribeToChangesResult", +] diff --git a/pubsub/live_event_subscriber.py b/pubsub/live_event_subscriber.py new file mode 100644 index 0000000..7e3aaa0 --- /dev/null +++ b/pubsub/live_event_subscriber.py @@ -0,0 +1,275 @@ +"""Subscribe to FastComments live changes over WebSocket. + +This mirrors the Java pubsub ``LiveEventSubscriber``: it opens a WebSocket to the +FastComments live server, hands each parsed :class:`LiveEvent` to a callback, fetches +any missed events from the event-log REST endpoint on (re)connect, and transparently +reconnects with jitter until the caller closes the subscription. +""" + +import json +import logging +import random +import threading +import time +from typing import Callable, Dict, List, Optional +from urllib.parse import urlencode + +from client import ApiClient, Configuration, PublicApi +from client.models.live_event import LiveEvent +from client.models.live_event_type import LiveEventType + +from .subscribe_to_changes_result import SubscribeToChangesResult + +try: + import websocket # provided by the "websocket-client" package +except ImportError: # pragma: no cover - surfaced lazily with a helpful message + websocket = None + +logger = logging.getLogger("fastcomments.pubsub") + +RECONNECT_INTERVAL_BASE_SECONDS = 4.0 +DEFAULT_PING_INTERVAL_SECONDS = 25 +TESTING_PING_INTERVAL_SECONDS = 5 +FETCH_EVENT_LOG_DEBOUNCE_SECONDS = 2.0 + +# Callback signatures: +# handle_live_event(event: LiveEvent) -> None +# can_see_comments(comment_ids: List[str]) -> Dict[str, str] (returns blocked-id map) +# on_connection_status_change(is_connected: bool, last_live_event_time: Optional[int]) -> None +HandleLiveEventCallback = Callable[[LiveEvent], None] +CanSeeCommentsCallback = Callable[[List[str]], Dict[str, str]] +ConnectionStatusChangeCallback = Callable[[bool, Optional[int]], None] + + +def _now_ms() -> int: + return int(time.time() * 1000) + + +class LiveEventSubscriber: + """Manages a live WebSocket subscription to FastComments events.""" + + def __init__(self, ping_interval_seconds: int = DEFAULT_PING_INTERVAL_SECONDS) -> None: + self._ping_interval_seconds = ping_interval_seconds + self.on_connection_status_change: Optional[ConnectionStatusChangeCallback] = None + + @classmethod + def create_testing(cls) -> "LiveEventSubscriber": + """Create a subscriber with a short ping interval, useful for aggressive NAT environments.""" + return cls(ping_interval_seconds=TESTING_PING_INTERVAL_SECONDS) + + @staticmethod + def _extract_comment_id_from_event(event: LiveEvent) -> Optional[str]: + if event.type == LiveEventType.NEW_MINUS_COMMENT and event.comment is not None: + return event.comment.id + return None + + def subscribe_to_changes( + self, + tenant_id_ws: str, + url_id: str, + url_id_ws: str, + user_id_ws: str, + handle_live_event: HandleLiveEventCallback, + can_see_comments: Optional[CanSeeCommentsCallback] = None, + on_connection_status_change: Optional[ConnectionStatusChangeCallback] = None, + region: Optional[str] = None, + disable_live_commenting: bool = False, + api_host: str = "https://fastcomments.com", + api_client: Optional[ApiClient] = None, + ) -> Optional[SubscribeToChangesResult]: + """Open a live subscription and return a handle for closing it. + + :param tenant_id_ws: The tenant id, sanitized for the websocket server. + :param url_id: The url id that events are tied to (used when fetching the event log). + :param url_id_ws: The url id, sanitized for the websocket server. + :param user_id_ws: The user's "presence id". + :param handle_live_event: Called with each :class:`LiveEvent` that is received. + :param can_see_comments: Optional callback that receives a list of comment ids and + returns a dict whose keys are the ids the user is NOT allowed to see. Events for + blocked comment ids are dropped before ``handle_live_event`` is called. + :param on_connection_status_change: Optional callback invoked with + ``(is_connected, last_live_event_time)`` whenever the connection opens or drops. + :param region: Set to ``"eu"`` to use the EU websocket host. + :param disable_live_commenting: When ``True``, no connection is opened and ``None`` is returned. + :param api_host: REST host used to fetch missed events from the event log. + :param api_client: Optional pre-configured :class:`client.ApiClient` for the event-log + requests. When omitted one is built from ``api_host``. + :return: A :class:`SubscribeToChangesResult`, or ``None`` if live commenting is disabled + or the ``websocket-client`` dependency is not installed. + """ + if disable_live_commenting: + return None + + if websocket is None: + logger.error( + "FastComments: the 'websocket-client' package is required for live subscriptions. " + "Install it with: pip install \"fastcomments[pubsub]\"" + ) + return None + + if on_connection_status_change is not None: + self.on_connection_status_change = on_connection_status_change + + if api_client is None: + # Default the event-log REST host from the region so the "fetch missed events" + # fallback queries EU infra for EU tenants instead of silently hitting US. + if api_host == "https://fastcomments.com" and region == "eu": + api_host = "https://eu.fastcomments.com" + api_client = ApiClient(Configuration(host=api_host)) + public_api = PublicApi(api_client) + + # Per-subscription state shared across reconnects. + state = { + "last_event_time": _now_ms(), + "intentionally_closed": False, + "app": None, # type: Optional[websocket.WebSocketApp] + "fetch_timer": None, # type: Optional[threading.Timer] + "reconnect_timer": None, # type: Optional[threading.Timer] + } + + ws_host = "wss://ws-eu.fastcomments.com" if region == "eu" else "wss://ws.fastcomments.com" + query = urlencode({"urlId": url_id_ws, "userIdWS": user_id_ws, "tenantIdWS": tenant_id_ws}) + ws_url = f"{ws_host}/sub?{query}" + + def process_events(events: List[LiveEvent]) -> None: + if not events: + return + + blocked_map: Optional[Dict[str, str]] = None + if can_see_comments is not None: + ids = [ + comment_id + for comment_id in (self._extract_comment_id_from_event(event) for event in events) + if comment_id is not None + ] + if ids: + try: + blocked_map = can_see_comments(ids) + except Exception as exc: # noqa: BLE001 - never let a user callback break the stream + logger.error("FastComments: can_see_comments callback failed: %s", exc) + blocked_map = None + + for event in events: + try: + if event.timestamp is not None: + state["last_event_time"] = max(state["last_event_time"], event.timestamp) + comment_id = self._extract_comment_id_from_event(event) + if blocked_map is None or comment_id not in blocked_map: + handle_live_event(event) + except Exception as exc: # noqa: BLE001 + logger.error("FastComments: error processing event: %s", exc) + + def fetch_event_log(start_time: int, end_time: int) -> None: + if state["intentionally_closed"]: + return + + def run() -> None: + try: + response = public_api.get_event_log( + tenant_id_ws, url_id, user_id_ws, start_time, end_time + ) + parsed: List[LiveEvent] = [] + for entry in response.events or []: + if entry.data: + try: + event = LiveEvent.from_json(entry.data) + if event is not None: + parsed.append(event) + except Exception as exc: # noqa: BLE001 + logger.error("FastComments: error parsing event data: %s", exc) + process_events(parsed) + except Exception as exc: # noqa: BLE001 + logger.error("FastComments: fetch_event_log failure: %s", exc) + + # Debounce so a burst of reconnect attempts doesn't hammer the event-log endpoint. + if state["fetch_timer"] is not None: + state["fetch_timer"].cancel() + timer = threading.Timer(FETCH_EVENT_LOG_DEBOUNCE_SECONDS, run) + timer.daemon = True + state["fetch_timer"] = timer + timer.start() + + def connect() -> None: + if state["intentionally_closed"]: + return + + def on_open(_app) -> None: + if state["intentionally_closed"]: + _app.close() + return + if state["last_event_time"] > 0: + fetch_event_log(state["last_event_time"], _now_ms()) + if self.on_connection_status_change is not None: + self.on_connection_status_change(True, state["last_event_time"]) + + def on_message(_app, message) -> None: + if state["intentionally_closed"]: + return + try: + event = LiveEvent.from_json(message) + if event is None: + return + if event.timestamp is not None: + # ts (present on presence-only updates) is intentionally not tracked, + # since those events are not stored in the log. + state["last_event_time"] = max(state["last_event_time"], event.timestamp) + process_events([event]) + except Exception as exc: # noqa: BLE001 + logger.error("FastComments: error processing WebSocket message: %s", exc) + + def on_error(_app, error) -> None: + logger.error("FastComments: WebSocket error: %s", error) + if state["last_event_time"] > 0: + fetch_event_log(state["last_event_time"], _now_ms()) + + def on_close(_app, _status_code, _msg) -> None: + if state["fetch_timer"] is not None: + state["fetch_timer"].cancel() + state["fetch_timer"] = None + + if state["last_event_time"] <= 0: + state["last_event_time"] = _now_ms() + + if state["intentionally_closed"]: + return + + if self.on_connection_status_change is not None: + self.on_connection_status_change(False, state["last_event_time"]) + + reconnect_delay = RECONNECT_INTERVAL_BASE_SECONDS * random.random() + reconnect_timer = threading.Timer(reconnect_delay, connect) + reconnect_timer.daemon = True + state["reconnect_timer"] = reconnect_timer + reconnect_timer.start() + + app = websocket.WebSocketApp( + ws_url, + on_open=on_open, + on_message=on_message, + on_error=on_error, + on_close=on_close, + ) + state["app"] = app + + def run_forever() -> None: + # Protocol-level ping keeps NATs (emulators, silly networks, etc) alive. + app.run_forever(ping_interval=self._ping_interval_seconds) + + thread = threading.Thread(target=run_forever, name="fastcomments-live", daemon=True) + thread.start() + + connect() + + def close() -> None: + state["intentionally_closed"] = True + if state["reconnect_timer"] is not None: + state["reconnect_timer"].cancel() + if state["fetch_timer"] is not None: + state["fetch_timer"].cancel() + if state["app"] is not None: + try: + state["app"].close() + except Exception as exc: # noqa: BLE001 + logger.error("FastComments: error closing WebSocket: %s", exc) + + return SubscribeToChangesResult(close) diff --git a/pubsub/subscribe_to_changes_result.py b/pubsub/subscribe_to_changes_result.py new file mode 100644 index 0000000..365532b --- /dev/null +++ b/pubsub/subscribe_to_changes_result.py @@ -0,0 +1,15 @@ +"""Result of subscribing to changes.""" + +from typing import Callable, Optional + + +class SubscribeToChangesResult: + """Handle returned by ``subscribe_to_changes``; call ``close()`` to stop the subscription.""" + + def __init__(self, close_action: Optional[Callable[[], None]] = None) -> None: + self._close_action = close_action + + def close(self) -> None: + """Intentionally close the connection and stop reconnecting.""" + if self._close_action is not None: + self._close_action() diff --git a/pyproject.toml b/pyproject.toml index d13a3a0..4e7e0e2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -34,8 +34,15 @@ client = [ "pydantic>=2.0.0", "typing-extensions>=4.0.0", ] +# Live subscriptions (pubsub.LiveEventSubscriber) need the generated client plus a +# WebSocket client: pip install "fastcomments[pubsub] @ git+..." +pubsub = [ + "fastcomments[client]", + "websocket-client>=1.5.0", +] dev = [ "fastcomments[client]", + "fastcomments[pubsub]", "pytest>=7.0.0", "pytest-cov>=4.0.0", "pytest-asyncio>=0.21.0", @@ -54,7 +61,7 @@ Issues = "https://github.com/fastcomments/fastcomments-python/issues" # packages = ["sso", "client"] omits them, which breaks non-editable installs # with "No module named 'client.api'". [tool.setuptools.packages.find] -include = ["sso*", "client*"] +include = ["sso*", "client*", "pubsub*"] [tool.setuptools.package-data] client = ["py.typed"] @@ -98,7 +105,7 @@ exclude = ''' ''' [tool.coverage.run] -source = ["sso"] +source = ["sso", "pubsub"] omit = [ "*/tests/*", "*/client/*", diff --git a/tests/test_pubsub.py b/tests/test_pubsub.py new file mode 100644 index 0000000..833f81d --- /dev/null +++ b/tests/test_pubsub.py @@ -0,0 +1,225 @@ +"""Unit tests for the live pubsub subscriber (no real network).""" + +import json +from unittest.mock import MagicMock, patch + +import pytest + +from client.models.event_log_entry import EventLogEntry +from client.models.get_event_log_response import GetEventLogResponse +from pubsub import LiveEventSubscriber, SubscribeToChangesResult + + +def _new_comment_event(comment_id: str, timestamp: int = 1000) -> dict: + """A minimal 'new-comment' LiveEvent payload as sent by the live server.""" + return { + "type": "new-comment", + "timestamp": timestamp, + "comment": { + "_id": comment_id, + "tenantId": "tenant-1", + "urlId": "url-1", + "commenterName": "Test User", + "commentHTML": "

hello

", + "comment": "hello", + "verified": True, + "url": "https://example.com/article", + "approved": True, + "date": "2024-01-01T00:00:00Z", + }, + } + + +class FakeWebSocketApp: + """Captures the handlers passed by the subscriber so tests can drive them synchronously.""" + + last_instance = None + + def __init__(self, url, on_open=None, on_message=None, on_error=None, on_close=None): + self.url = url + self.on_open = on_open + self.on_message = on_message + self.on_error = on_error + self.on_close = on_close + self.closed = False + FakeWebSocketApp.last_instance = self + + def run_forever(self, ping_interval=None): + # Do not block; a real run loop would sit here until close. + self.ping_interval = ping_interval + + def close(self): + self.closed = True + + +@pytest.fixture(autouse=True) +def fake_websocket(): + with patch("pubsub.live_event_subscriber.websocket") as ws_module: + ws_module.WebSocketApp = FakeWebSocketApp + FakeWebSocketApp.last_instance = None + yield ws_module + + +def _make_subscriber(): + """A subscriber whose event-log fetch is a no-op unless a test overrides it.""" + return LiveEventSubscriber() + + +def test_disable_live_commenting_returns_none(): + result = LiveEventSubscriber().subscribe_to_changes( + "tenant-1", "url-1", "url-1", "user-ws-1", lambda e: None, disable_live_commenting=True + ) + assert result is None + + +def test_returns_result_and_builds_ws_url(): + result = _make_subscriber().subscribe_to_changes( + "tenant-1", "url-1", "url-1-ws", "user-ws-1", lambda e: None + ) + assert isinstance(result, SubscribeToChangesResult) + app = FakeWebSocketApp.last_instance + assert app is not None + assert app.url.startswith("wss://ws.fastcomments.com/sub?") + assert "urlId=url-1-ws" in app.url + assert "userIdWS=user-ws-1" in app.url + assert "tenantIdWS=tenant-1" in app.url + result.close() + + +def test_eu_region_uses_eu_host(): + _make_subscriber().subscribe_to_changes( + "tenant-1", "url-1", "url-1", "user-ws-1", lambda e: None, region="eu" + ) + assert FakeWebSocketApp.last_instance.url.startswith("wss://ws-eu.fastcomments.com/sub?") + + +def test_on_message_dispatches_parsed_live_event(): + received = [] + _make_subscriber().subscribe_to_changes( + "tenant-1", "url-1", "url-1", "user-ws-1", received.append + ) + app = FakeWebSocketApp.last_instance + app.on_message(app, json.dumps(_new_comment_event("comment-1"))) + assert len(received) == 1 + assert received[0].type.value == "new-comment" + assert received[0].comment.id == "comment-1" + + +def test_can_see_comments_blocks_events(): + received = [] + _make_subscriber().subscribe_to_changes( + "tenant-1", + "url-1", + "url-1", + "user-ws-1", + received.append, + can_see_comments=lambda ids: {"comment-blocked": "1"}, + ) + app = FakeWebSocketApp.last_instance + app.on_message(app, json.dumps(_new_comment_event("comment-blocked"))) + app.on_message(app, json.dumps(_new_comment_event("comment-visible"))) + assert [e.comment.id for e in received] == ["comment-visible"] + + +def test_connection_status_callback_fires_on_open(): + statuses = [] + _make_subscriber().subscribe_to_changes( + "tenant-1", + "url-1", + "url-1", + "user-ws-1", + lambda e: None, + on_connection_status_change=lambda connected, ts: statuses.append(connected), + ) + app = FakeWebSocketApp.last_instance + app.on_open(app) + assert statuses == [True] + + +def test_close_marks_intentional_and_prevents_reconnect(): + result = _make_subscriber().subscribe_to_changes( + "tenant-1", "url-1", "url-1", "user-ws-1", lambda e: None + ) + app = FakeWebSocketApp.last_instance + result.close() + assert app.closed is True + # An intentional close must not schedule a reconnect (no new app instance). + app.on_close(app, 1000, "bye") + assert FakeWebSocketApp.last_instance is app + + +def test_eu_region_event_log_uses_eu_host(): + with patch("pubsub.live_event_subscriber.ApiClient") as ApiClientMock, patch( + "pubsub.live_event_subscriber.Configuration" + ) as ConfigurationMock: + _make_subscriber().subscribe_to_changes( + "tenant-1", "url-1", "url-1", "user-ws-1", lambda e: None, region="eu" + ) + ConfigurationMock.assert_called_once_with(host="https://eu.fastcomments.com") + + +def test_us_region_event_log_uses_us_host(): + with patch("pubsub.live_event_subscriber.ApiClient"), patch( + "pubsub.live_event_subscriber.Configuration" + ) as ConfigurationMock: + _make_subscriber().subscribe_to_changes( + "tenant-1", "url-1", "url-1", "user-ws-1", lambda e: None + ) + ConfigurationMock.assert_called_once_with(host="https://fastcomments.com") + + +def test_explicit_api_host_overrides_region_default(): + with patch("pubsub.live_event_subscriber.ApiClient"), patch( + "pubsub.live_event_subscriber.Configuration" + ) as ConfigurationMock: + _make_subscriber().subscribe_to_changes( + "tenant-1", + "url-1", + "url-1", + "user-ws-1", + lambda e: None, + region="eu", + api_host="https://my-proxy.example.com", + ) + ConfigurationMock.assert_called_once_with(host="https://my-proxy.example.com") + + +def test_open_fetches_missed_events_from_event_log(): + received = [] + subscriber = _make_subscriber() + + missed = GetEventLogResponse( + events=[ + EventLogEntry( + _id="e1", + createdAt="2024-01-01T00:00:00Z", + tenantId="tenant-1", + urlId="url-1", + broadcastId="b1", + data=json.dumps(_new_comment_event("missed-comment", timestamp=999)), + ) + ], + status="success", + ) + + with patch("pubsub.live_event_subscriber.PublicApi") as PublicApiMock, patch( + "pubsub.live_event_subscriber.FETCH_EVENT_LOG_DEBOUNCE_SECONDS", 0.0 + ): + instance = PublicApiMock.return_value + instance.get_event_log.return_value = missed + + result = subscriber.subscribe_to_changes( + "tenant-1", "url-1", "url-1", "user-ws-1", received.append + ) + app = FakeWebSocketApp.last_instance + app.on_open(app) + # Debounced fetch runs on a Timer thread; give it a moment. + import time + + deadline = time.time() + 2 + while not received and time.time() < deadline: + time.sleep(0.02) + + result.close() + assert instance.get_event_log.called + assert [e.comment.id for e in received] == ["missed-comment"]