From 2bc2b0c7f0c0c462c2cf0f4fb205cbb6b0f90c1c Mon Sep 17 00:00:00 2001 From: EricPaque <60603143+Radisio@users.noreply.github.com> Date: Mon, 17 Aug 2026 18:18:12 +0200 Subject: [PATCH] Include SSE instead of pooling --- .env.exemple | 8 ++ api/administrative_document/service.py | 10 ++ core/config.py | 8 ++ core/notifications/service.py | 47 ++++++++ core/realtime/__init__.py | 54 +++++++++ core/realtime/bus.py | 154 +++++++++++++++++++++++++ core/realtime/channels.py | 89 ++++++++++++++ core/realtime/envelope.py | 92 +++++++++++++++ main.py | 4 + requirements/base.txt | 7 +- worker/main.py | 4 + 11 files changed, 476 insertions(+), 1 deletion(-) create mode 100644 core/realtime/__init__.py create mode 100644 core/realtime/bus.py create mode 100644 core/realtime/channels.py create mode 100644 core/realtime/envelope.py diff --git a/.env.exemple b/.env.exemple index a506b35..eeb82f5 100644 --- a/.env.exemple +++ b/.env.exemple @@ -55,6 +55,14 @@ RENDER_STALE_AFTER_SECONDS=1800 # (mount the file and set this) — an unresolvable registry fails every dossier. REGULATORS_CONFIG_PATH= +# ── 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=* diff --git a/api/administrative_document/service.py b/api/administrative_document/service.py index f7e64fa..5e63362 100644 --- a/api/administrative_document/service.py +++ b/api/administrative_document/service.py @@ -786,6 +786,11 @@ async def _transition( "context": clean_context, }, ) + # _notify_acknowledged() only STAGES; _audit_and_commit() above owns the + # CRM commit, so this is the first post-commit point. It swallows commit + # failures, so we may flush after a rollback: harmless, because the + # client refetches, finds no row, and the unread count does not move. + await self._notify.flush_realtime() # ------------------------------------------------------------------ # Deadline derivation @@ -1041,6 +1046,11 @@ async def sweep_deadlines(self, today: datetime.date | None = None) -> tuple[int for deadline in overdue: await self._publish_deadline(SUBJECT_DEADLINE_MISSED, EVENT_DEADLINE_MISSED, deadline) + # ONE flush, here, after self._local.commit() — not two, and not after + # the CRM commits above. `reminded_at` and the MISSED statuses are the + # business write and they commit LAST, so flushing earlier would tell the + # browser to refetch pre-commit state with no second event coming. + await self._notify.flush_realtime() return len(overdue), rolled # ------------------------------------------------------------------ diff --git a/core/config.py b/core/config.py index 9d4b25b..3b619db 100644 --- a/core/config.py +++ b/core/config.py @@ -99,6 +99,14 @@ class Settings(BaseSettings): # (mount the file and set this). REGULATORS_CONFIG_PATH: str = "" + # ---- 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 = "*" diff --git a/core/notifications/service.py b/core/notifications/service.py index bcb08cc..9911ef7 100644 --- a/core/notifications/service.py +++ b/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: diff --git a/core/realtime/__init__.py b/core/realtime/__init__.py new file mode 100644 index 0000000..5606326 --- /dev/null +++ b/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", +] diff --git a/core/realtime/bus.py b/core/realtime/bus.py new file mode 100644 index 0000000..8abecc5 --- /dev/null +++ b/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 diff --git a/core/realtime/channels.py b/core/realtime/channels.py new file mode 100644 index 0000000..2144039 --- /dev/null +++ b/core/realtime/channels.py @@ -0,0 +1,89 @@ +"""THE ONLY PLACE A REALTIME CHANNEL STRING IS BUILT (Python side). + +This is a security control, not a style rule. A producer that accidentally +publishes a per-user thing onto a community tier is a cross-tenant leak, and +``tier`` below is a required argument with no default precisely so that mistake +cannot be made by omission. A test greps each service for the literal +``notify:v1:`` outside this module and fails on a hit. + +Grammar (fixed arity, so Redis 6 ACLs can later be granted per prefix without a +redesign) — byte-identical to +``crm-backend/src/shared/realtime/realtime.channels.ts``:: + + notify:v1:u:{internal_app_user_id} + notify:v1:c:{internal_community_id}:{MEMBER|MANAGER} + +Ids are the INTERNAL integer keys (``app_user.id``, ``community.id``) — what +every producer here already holds, and what ``notification.id_user`` is. They are +NOT Keycloak subs or org uuids. + +The community family is what lets a worker with NO user attribution at all (the +generation and simulation jobs carry only ``id_community``) address exactly the +right audience with zero database lookups. Its safety comes from the subscribe +side: crm-backend only ever subscribes a connection to tiers the ticket mint +proved the user holds, from gateway-verified claims. +""" + +from collections.abc import Iterable, Sequence +from dataclasses import dataclass +from enum import Enum + +_PREFIX = "notify:v1" + + +class Tier(str, Enum): + """Channel tiers. Deliberately coarser than a role: there is no ADMIN bucket.""" + + #: Everyone in the community, managers included. + MEMBER = "MEMBER" + #: ADMIN and MANAGER only. + MANAGER = "MANAGER" + + +def user_channel(internal_user_id: int) -> str: + """Channel for one user, in every community and outside all of them.""" + return f"{_PREFIX}:u:{internal_user_id}" + + +def community_channel(internal_community_id: int, tier: Tier) -> str: + """Channel for one tier of one community.""" + return f"{_PREFIX}:c:{internal_community_id}:{tier.value}" + + +@dataclass(frozen=True) +class UserAudience: + """One recipient, addressed by internal ``app_user.id``.""" + + user_id: int + + def channels(self) -> Sequence[str]: + return (user_channel(self.user_id),) + + +@dataclass(frozen=True) +class UsersAudience: + """An explicit set of recipients. Duplicates are collapsed.""" + + user_ids: Iterable[int] + + def channels(self) -> Sequence[str]: + return tuple(user_channel(uid) for uid in dict.fromkeys(self.user_ids)) + + +@dataclass(frozen=True) +class CommunityAudience: + """One tier of one community. + + ``Tier.MANAGER`` reaches ADMIN and MANAGER only; ``Tier.MEMBER`` reaches + everyone in the community, managers included — a manager's connection + subscribes to both tiers, so "everyone" is one publish, not two. + """ + + community_id: int + tier: Tier + + def channels(self) -> Sequence[str]: + return (community_channel(self.community_id, self.tier),) + + +Audience = UserAudience | UsersAudience | CommunityAudience diff --git a/core/realtime/envelope.py b/core/realtime/envelope.py new file mode 100644 index 0000000..a2db705 --- /dev/null +++ b/core/realtime/envelope.py @@ -0,0 +1,92 @@ +"""Realtime envelope construction and validation. + +An envelope is a HINT — "something about resource X changed" — never data. The +rules below mirror ``crm-backend/src/shared/realtime/realtime.envelope.ts``, +which re-validates everything on the way out to a browser: + +* **No business data.** No names, emails, EANs, amounts, invoice numbers, + storage keys, error messages. +* **No display strings.** Toast text is chosen client-side from ``topic`` + + ``hint["status"]`` against the i18n bundle. This is a security control: a + compromised publisher gets a nuisance channel, never a text-injection channel + into every open browser. +* **No recipient field.** The channel already says who. A recipient in the body + invites a subscriber-side "is this for me?" check — authorization on the wrong + leg. + +``ref.id`` is permitted: any authorized reader can already see it, and the +client needs it to decide *which* row to refetch. +""" + +import json +import secrets +from collections.abc import Mapping +from datetime import UTC, datetime +from typing import Any, Final + +#: Hard ceiling on a serialized envelope, in BYTES (not characters). +MAX_ENVELOPE_BYTES: Final[int] = 1024 + +#: The topic registry. Mirrors crm-backend's realtime.topics.ts; an unknown topic +#: is dropped by the hub rather than forwarded, so publishing one is a silent +#: no-op that is much easier to find here. +TOPICS: Final[frozenset[str]] = frozenset( + { + "notification.created", + "generation.finished", + "simulation.finished", + "billing_run.finished", + "session.revoked", + } +) + +_SCALARS = (str, int, float, bool) + + +def _is_scalar(value: Any) -> bool: + # bool is a subclass of int, so it is already covered; None is allowed. + return value is None or isinstance(value, _SCALARS) + + +def build_envelope( + *, + topic: str, + resource: tuple[str, str | int], + hint: Mapping[str, str | int | float | bool | None] | None = None, + scope_community_id: int | None = None, +) -> dict[str, Any] | None: + """Build a valid envelope, or ``None`` if the input violates the contract. + + Returns ``None`` rather than raising: every caller is a fire-and-forget side + effect that must never affect a business write, so a malformed hint has to + degrade to "no event", not to an exception travelling up through a commit + path. + """ + if topic not in TOPICS: + return None + + kind, ref_id = resource + if not kind or ref_id is None: + return None + + flat: dict[str, Any] = {} + for key, value in (hint or {}).items(): + if not _is_scalar(value): + return None + flat[str(key)] = value + + envelope: dict[str, Any] = { + "v": 1, + # A client-side dedupe key. Not sortable on purpose: there is no replay, + # so ordering buys nothing. + "id": secrets.token_hex(8), + "topic": topic, + "at": datetime.now(UTC).isoformat().replace("+00:00", "Z"), + "scope": {"community_id": scope_community_id}, + "ref": {"kind": str(kind), "id": str(ref_id)}, + "hint": flat, + } + + if len(json.dumps(envelope, separators=(",", ":")).encode("utf-8")) > MAX_ENVELOPE_BYTES: + return None + return envelope diff --git a/main.py b/main.py index 530c7f6..2a28c93 100644 --- a/main.py +++ b/main.py @@ -16,6 +16,7 @@ from core.middleware.request_limits import RequestLimitsMiddleware from core.middleware.set_auth_context import GatewayScopeMiddleware from core.queue.init import close_nats, init_nats +from core.realtime import log_realtime_state from core.tracing import enrich_span, setup_tracer_provider configure_logging() @@ -24,6 +25,9 @@ @asynccontextmanager async def lifespan(app: FastAPI): + # Absence of this line means the image predates the realtime feature — + # see core/realtime/bus.py. Must come after configure_logging(). + log_realtime_state("administrative-document api") setup_tracer_provider() await init_nats() yield diff --git a/requirements/base.txt b/requirements/base.txt index e87a86f..46d64e1 100644 --- a/requirements/base.txt +++ b/requirements/base.txt @@ -37,4 +37,9 @@ opentelemetry-util-http==0.60b1 # "+10 business days" never drifts; dateutil.relativedelta does calendar-month # offsets ("+6 months", "+1 year"). See domain/calendar.py. workalendar==17.0.0 -python-dateutil==2.9.0.post0 \ No newline at end of file +python-dateutil==2.9.0.post0 + +# Realtime SSE hints (fire-and-forget pub/sub consumed by crm-backend's hub). +# redis.asyncio, not the unmaintained aioredis — it is aioredis's successor and +# ships in the same package. +redis==5.2.1 diff --git a/worker/main.py b/worker/main.py index ee867ae..2987dc8 100644 --- a/worker/main.py +++ b/worker/main.py @@ -24,6 +24,7 @@ from core.database.database import crm_engine, local_engine from core.logging import configure_logging from core.queue.init import close_nats, get_jetstream, init_nats +from core.realtime import log_realtime_state from core.tracing import setup_tracer_provider from worker import dispatcher from worker.scheduler import run_deadline_scheduler @@ -92,6 +93,9 @@ async def _heartbeat(shutdown_event: asyncio.Event) -> None: async def main() -> None: configure_logging() + # Absence of this line means the image predates the realtime feature — + # see core/realtime/bus.py. Must come after configure_logging(). + log_realtime_state("administrative-document-worker") setup_tracer_provider() await _connect_nats_with_retry()