Features/sse

.env.exemple

@@ -56,6 +56,14 @@ INVOICE_PROFORMA_TEMPLATE_URI=s3://optimce-templates/billing/invoice_proforma/v1
 # ── Localization ──
 DEFAULT_LOCALE=fr-BE
 
+# ── Realtime (Redis pub/sub) ──
+# Fire-and-forget SSE hints consumed by crm-backend's realtime hub. Optional
+# everywhere, including production: an empty URL makes core.realtime.emit() a
+# silent no-op, so a broker outage degrades the UI to polling and never affects
+# a business write. db 1 on purpose — db 0 is crm-backend's cache.
+REALTIME_ENABLED=false
+REALTIME_REDIS_URL=redis://:changeme_redis_password@localhost:6379/1
+
 # ── CORS ──
 ALLOW_ORIGIN=*
 

api/billing/service.py

@@ -392,6 +392,7 @@ async def issue_invoice(self, *, invoice_id: int) -> IssueOut:
         )
         await self._crm.commit()
         app_metrics.invoices_issued.add(1)
+        await self._notify.flush_realtime()
 
         return IssueOut(
             id=invoice_id,
@@ -541,6 +542,12 @@ async def sweep_overdue(self) -> OverdueSweepOut:
             await self._crm.commit()
         await self._local.commit()
         return OverdueSweepOut(marked=len(swept))
+        # AFTER self._local.commit(), not after the CRM one above: the
+        # invoice -> OVERDUE UPDATE is the row the UI refetches, and it is the
+        # local commit that makes it durable. Note the two commits are ordered
+        # the OPPOSITE way round in issue_invoice, so "flush after the enclosing
+        # commit" is not a rule that survives being applied mechanically.
+        await self._notify.flush_realtime()
 
     async def create_credit_note(self, *, invoice_id: int, body: CreditNoteIn) -> InvoiceOut:
         cid = self._community()

core/config.py

@@ -61,6 +61,14 @@ class Settings(BaseSettings):
     # MinIO ignores region but botocore still requires it to sign requests.
     STORAGE_REGION: str = "us-east-1"
 
+    # ---- Realtime (Redis pub/sub) ----
+    # Fire-and-forget SSE hints. Deliberately NOT in validate_env_config's
+    # required set (contrast NATS_URL below): realtime is optional by design, and
+    # making it mandatory would turn a broker outage into a boot failure. An
+    # empty URL makes core.realtime.emit() a silent no-op.
+    REALTIME_ENABLED: bool = False
+    REALTIME_REDIS_URL: str = ""
+
     # ---- CORS ----
     ALLOW_ORIGIN: str = "*"
 

core/notifications/service.py

@@ -27,6 +27,7 @@
 )
 from core.notifications.dedupe import build_dedupe_key, type_prefix_of
 from core.notifications.repository import NotificationRepository
+from core.realtime import UsersAudience, emit
 from shared.models.crm_models import NotificationModel
 
 logger = logging.getLogger(__name__)
@@ -56,6 +57,10 @@ class NotificationService:
     def __init__(self, crm_session: AsyncSession):
         self.crm_session = crm_session
         self.repository = NotificationRepository(crm_session)
+        # Realtime hints staged by publish() and released by flush_realtime()
+        # AFTER the caller commits. See flush_realtime's docstring for why this
+        # cannot simply be emitted inline.
+        self._pending_realtime: list[tuple[list[int], int | None]] = []
 
     async def publish(
         self,
@@ -151,6 +156,16 @@ async def publish(
                             for user_id in email_ids
                         ],
                     )
+            # Stage the realtime hint. A list append cannot abort a Postgres
+            # transaction, and this sits outside the savepoint above, so a
+            # savepoint rollback re-raises into the `except` below and nothing is
+            # staged — no event for rows that never landed.
+            #
+            # The audience is the SAME inapp_ids resolution performed above, never
+            # a second one: a divergence there means a notification row with no
+            # event, or an event with no row.
+            if inapp_ids:
+                self._pending_realtime.append((list(inapp_ids), community_id))
             return len(rows)
         except Exception:
             logger.exception(
@@ -159,6 +174,38 @@ async def publish(
             )
             return 0
 
+    async def flush_realtime(self) -> None:
+        """Release the realtime hints staged by publish(). NEVER raises.
+
+        *** CALL THIS AFTER YOUR COMMIT. ***
+
+        publish() deliberately runs INSIDE the caller's transaction because it
+        writes rows. A realtime hint has the exact opposite requirement: emitted
+        before the commit, it tells the browser to refetch and read PRE-COMMIT
+        state, and because the transport is fire-and-forget there is no second
+        event — a permanently stale UI behind a 200. So publish() is split in two
+        and the caller, which owns the commit, owns the boundary between them.
+
+        Not wired to SQLAlchemy's ``after_commit`` event on purpose: that fires
+        synchronously inside the greenlet, so an async publish there needs
+        ``create_task``, and in a worker that exits immediately the task can be
+        garbage-collected before it ever runs.
+
+        Safe to call when nothing is staged, and safe to call twice.
+        """
+        staged, self._pending_realtime = self._pending_realtime, []
+        for user_ids, community_id in staged:
+            await emit(
+                topic="notification.created",
+                audience=UsersAudience(user_ids=user_ids),
+                # The client refetches /unread-count and the recent slice, so it
+                # needs neither a row id nor a count — and the envelope must
+                # carry no business data regardless.
+                resource=("notification", "0"),
+                scope_community_id=community_id,
+                hint={},
+            )
+
     async def _resolve_audience(self, target: NotificationTarget) -> tuple[list[int], int | None]:
         """Turn a target into (de-duplicated recipient ids, source community)."""
         match target:

core/realtime/__init__.py

@@ -0,0 +1,54 @@
+"""Shared realtime publisher — BYTE-IDENTICAL across every producing service.
+
+Copied verbatim into billing, administrative-document, news-board,
+allocation-key-generation and simulation-key, mirroring the ``core/notifications``
+convention. ``scripts/check-realtime-parity.sh`` at the monorepo root is the gate:
+make the change in the reference service (news-board) and copy it out, never edit
+one copy.
+
+Consumed by crm-backend's realtime hub (``src/shared/realtime/``) and delivered
+to browsers over SSE. Fire-and-forget by contract: if the recipient has no stream
+open the event is dropped, which is correct — every event is a hint, and the
+client refetches authoritative state through the API gateway.
+
+Usage, always AFTER the owning transaction commits::
+
+    from core.realtime import CommunityAudience, Tier, emit
+
+    await session.commit()
+    await emit(
+        topic="generation.finished",
+        audience=CommunityAudience(community_id=cid, tier=Tier.MANAGER),
+        resource=("generation", generation_id),
+        scope_community_id=cid,
+        hint={"status": "success"},
+    )
+"""
+
+from .bus import close, emit, log_realtime_state
+from .channels import (
+    Audience,
+    CommunityAudience,
+    Tier,
+    UserAudience,
+    UsersAudience,
+    community_channel,
+    user_channel,
+)
+from .envelope import MAX_ENVELOPE_BYTES, TOPICS, build_envelope
+
+__all__ = [
+    "MAX_ENVELOPE_BYTES",
+    "TOPICS",
+    "Audience",
+    "CommunityAudience",
+    "Tier",
+    "UserAudience",
+    "UsersAudience",
+    "build_envelope",
+    "close",
+    "community_channel",
+    "emit",
+    "log_realtime_state",
+    "user_channel",
+]

core/realtime/bus.py

@@ -0,0 +1,154 @@
+"""Fire-and-forget realtime publisher.
+
+One lazily-created ``redis.asyncio`` client per process. Every failure mode —
+no configuration, unreachable broker, hung broker, malformed envelope — is a
+silent no-op, because the *only* thing lost is UI freshness: crm-backend's
+clients refetch authoritative state on every reconnect, and every poller in the
+SPA keeps running (slower) as a durability backstop.
+"""
+
+import asyncio
+import json
+import logging
+from collections.abc import Mapping
+
+import redis.asyncio as redis
+
+from core.config import settings
+
+from .channels import Audience
+from .envelope import build_envelope
+
+logger = logging.getLogger(__name__)
+
+_client: redis.Redis | None = None
+
+#: Publishing must never add latency to a request or a worker tick. A wedged
+#: broker is bounded here rather than by the socket, because a *connected but
+#: hung* Redis would otherwise await forever.
+_PUBLISH_TIMEOUT_SECONDS = 1.0
+
+
+def _redacted_url() -> str:
+    """The DSN with its password removed. NEVER log the raw value.
+
+    ``REALTIME_REDIS_URL`` is composed from ``REDIS_PASSWORD`` in
+    docker-compose, so it carries a live secret in userinfo position.
+    """
+    url = settings.REALTIME_REDIS_URL
+    scheme, sep, rest = url.partition("://")
+    if not sep or "@" not in rest:
+        return url
+    return f"{scheme}://***@{rest.rpartition('@')[2]}"
+
+
+def log_realtime_state(component: str) -> None:
+    """Announce this process's realtime publishing state, once, at startup.
+
+    *** CALL THIS AFTER configure_logging(). NEVER at module import time. ***
+
+    This module is imported long before logging is configured: ``worker/main.py``
+    imports the dispatcher, which reaches ``worker/persistence.py``, which imports
+    this file — all in the import block at the top — while ``configure_logging()``
+    runs inside ``async def main()``. And ``core/logging.py`` opens with
+    ``root.handlers.clear()``. So at import time the root logger has no handlers,
+    Python falls back to ``logging.lastResort`` at WARNING, and an INFO line here
+    is dropped with no trace whatsoever — reproducing the exact silence this
+    function exists to break. Do not "simplify" it into the module body.
+
+    Why it exists: four worker containers once ran images built before this
+    package existed. ``core/realtime/`` was absent and so were the emit call
+    sites, so three topics were published by nothing at all — for three days,
+    while their environment variables looked perfectly correct, because
+    ``--force-recreate`` rebuilds the container from the EXISTING image. The
+    absence of this line is the cheapest signal that an image predates the
+    feature. ``scripts/check-realtime-images.sh`` is the automatable form of the
+    same check, and ``scripts/check-realtime-parity.sh`` cannot see it at all —
+    it compares source trees, which were green throughout.
+    """
+    if not settings.REALTIME_ENABLED:
+        logger.info("Realtime disabled — %s publishes nothing", component)
+        return
+    if not settings.REALTIME_REDIS_URL:
+        # Enabled but unconfigured is a misconfiguration, not a deployment choice.
+        logger.warning(
+            "Realtime ENABLED but REALTIME_REDIS_URL is empty — %s publishes nothing", component
+        )
+        return
+    logger.info("Realtime publisher ready — %s publishing to %s", component, _redacted_url())
+
+
+def _get_client() -> redis.Redis:
+    global _client
+    if _client is None:
+        _client = redis.from_url(
+            settings.REALTIME_REDIS_URL,
+            socket_connect_timeout=1,
+            socket_timeout=1,
+            health_check_interval=30,
+            decode_responses=False,
+        )
+    return _client
+
+
+async def emit(
+    *,
+    topic: str,
+    audience: Audience,
+    resource: tuple[str, str | int],
+    hint: Mapping[str, str | int | float | bool | None] | None = None,
+    scope_community_id: int | None = None,
+) -> None:
+    """Publish a realtime hint. NEVER raises.
+
+    *** CALL THIS AFTER THE COMMIT. NEVER inside ``begin_nested()``, and never
+    inside the ``try`` that owns the business write. ***
+
+    Publishing pre-commit does not merely lose an event — it tells the browser to
+    refetch and read PRE-COMMIT state, and because the transport is
+    fire-and-forget there is NO second event, ever. The result is a permanently
+    stale UI behind a 200, with no error anywhere. That is the same silhouette as
+    the sweep-commit-ordering and notification-savepoint traps.
+
+    Note the deliberate asymmetry with ``core.notifications.service.publish()``,
+    which MUST run inside the caller's transaction because it writes rows. These
+    two have opposite requirements. Do not "unify" them.
+
+    Do not fire this as a bare ``asyncio.create_task`` either: in a worker that
+    finishes immediately the task is orphaned (the event is lost anyway) and may
+    log after teardown.
+    """
+    if not settings.REALTIME_ENABLED or not settings.REALTIME_REDIS_URL:
+        return  # no-op: the default everywhere realtime is not deployed
+
+    try:
+        envelope = build_envelope(
+            topic=topic,
+            resource=resource,
+            hint=hint,
+            scope_community_id=scope_community_id,
+        )
+        if envelope is None:
+            logger.warning("realtime: envelope rejected locally topic=%s", topic)
+            return
+
+        body = json.dumps(envelope, separators=(",", ":"))
+        client = _get_client()
+        async with asyncio.timeout(_PUBLISH_TIMEOUT_SECONDS):
+            for channel in audience.channels():
+                await client.publish(channel, body)
+    # Blanket by contract: nothing this function can hit is worth propagating
+    # into a caller that has already committed.
+    except Exception:
+        logger.warning("realtime: emit failed topic=%s", topic, exc_info=True)
+
+
+async def close() -> None:
+    """Release the client. For worker shutdown and test teardown."""
+    global _client
+    if _client is not None:
+        try:
+            await _client.aclose()
+        except Exception:
+            logger.warning("realtime: client close failed", exc_info=True)
+        _client = None