diff --git a/CLAUDE.md b/CLAUDE.md index 54fe9cd..6d61a1e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -6,9 +6,10 @@ the domain. Full plan: `~/.claude/plans/snappy-mixing-dahl.md`. ## Layout - `api/billing/` — `routes` → `service` (orchestration) → `repository` (owned DB) + `mappers`/`schemas`/`deps`. -- `ports/` — `crm_core*` (read-only CRM adapter), `document_generation*` (async NATS), `email*` (Noop), `events` (NATS publish). +- `ports/` — `crm_core*` (read-only CRM adapter), `document_generation*` (async NATS), `email*` (Noop), `events` (NATS publish), `providers` (which adapter backs each port — framework-free on purpose, see `worker/` below). - `regime/` — `BillingRegime` Protocol + `CwapeWalloniaRegime` + `registry` (startup parity gate); config from `reference/regulators.json` + `regime/billing_regimes.json`. -- `worker/` — `dispatcher` (consumers) → `persistence.process_billing_run`, `issue.process_issue`, `docgen_results.process_docgen_result` (each callable directly for tests via injected sessions). +- `worker/` — `dispatcher` (consumers) → `persistence.process_billing_run`, `issue.process_issue`, `docgen_results.process_docgen_result` (each callable directly for tests via injected sessions); `scheduler`/`sweeps` = the 06:00 Brussels overdue tick. + - **`Dockerfile.worker` has NO fastapi/uvicorn/starlette** (it installs `requirements/worker.txt`). The worker may import `api.*` — that package IS copied, for `repository`/`mappers`/`BillingService` — but never `api/billing/deps.py` or anything else pulling the HTTP stack: get providers from `ports/providers.py`. This is invisible locally, where `.venv` has everything; it shipped once and crash-looped 62 times. `tests/worker/test_worker_import_graph.py` pins both the import graph and the Dockerfile's `COPY` list. - `utils/` — `money` (Decimal HALF_UP), `ogm` (mod-97), `numbering`. - `scripts/sql/schema.sql` — raw DDL (NO Alembic). `shared/models/local_models.py` mirrors it. diff --git a/Dockerfile.worker b/Dockerfile.worker index 5916ece..ff6e86f 100644 --- a/Dockerfile.worker +++ b/Dockerfile.worker @@ -27,10 +27,21 @@ WORKDIR /app # Copy only installed packages from builder COPY --from=builder /install /usr/local -# Copy application code. The worker imports api/ (repository + docgen payload -# builder), so it is included alongside the domain/infra packages. No main.py — -# the worker has its own entrypoint. Set REGULATORS_CONFIG_PATH to a mounted -# reference/regulators.json (outside this build context) for the parity check. +# Copy application code. The worker imports api/ (repository, docgen payload +# builder, and BillingService — the overdue sweep runs the real service so the +# route and the daily tick share one implementation), so it is included +# alongside the domain/infra packages. No main.py — the worker has its own +# entrypoint. +# +# api/ is here as a PACKAGE, not as the HTTP layer: this image installs +# requirements/worker.txt, which has no fastapi/uvicorn, so any worker module +# that reaches something importing fastapi kills the container at startup with +# ModuleNotFoundError. That is why the port providers live in ports/providers.py +# rather than api/billing/deps.py, and it is pinned by +# tests/worker/test_worker_import_graph.py. +# +# Set REGULATORS_CONFIG_PATH to a mounted reference/regulators.json (outside +# this build context) for the parity check. COPY core/ core/ COPY api/ api/ COPY shared/ shared/ diff --git a/api/billing/deps.py b/api/billing/deps.py index e1e4333..280f496 100644 --- a/api/billing/deps.py +++ b/api/billing/deps.py @@ -12,17 +12,15 @@ from core.database.database import get_crm_session, get_local_session from ports.crm_core_sqlalchemy import SqlAlchemyCrmCoreRead from ports.email import EmailPort -from ports.email_noop import NoopEmailAdapter -from ports.events import EventPublisher, NatsEventPublisher -from regime.registry import get_registry - - -def get_event_publisher() -> EventPublisher: - return NatsEventPublisher() +from ports.events import EventPublisher +# Which adapter backs each port is decided in `ports/providers.py`, not here, so +# the worker can make the same choice without importing fastapi. Re-exported +# because `dependency_overrides[deps.get_event_publisher]` keys on this object. +from ports.providers import get_email_port, get_event_publisher +from regime.registry import get_registry -def get_email_port() -> EmailPort: - return NoopEmailAdapter() +__all__ = ["get_billing_service", "get_email_port", "get_event_publisher"] def get_billing_service( diff --git a/api/billing/repository.py b/api/billing/repository.py index a95f53e..b7d8809 100644 --- a/api/billing/repository.py +++ b/api/billing/repository.py @@ -10,6 +10,7 @@ from __future__ import annotations import datetime +from dataclasses import dataclass from decimal import Decimal from typing import Any @@ -42,6 +43,19 @@ } +@dataclass(frozen=True, slots=True) +class SweptInvoice: + """The identity of one invoice the overdue sweep flipped. + + Just enough to address a notification at the member it bills — the sweep is a + bulk UPDATE and never loads the ORM instances. + """ + + id: int + id_member: int + number: str | None + + def _invoice_order_by(sort: str | None, order: str | None) -> list[Any]: """Build a safe ORDER BY for the invoice list endpoints. @@ -348,8 +362,15 @@ async def mark_paid(self, invoice_id: int, paid_at: datetime.datetime) -> None: .values(status=InvoiceStatus.PAID, paid_at=paid_at) ) - async def sweep_overdue(self, today: datetime.date) -> int: - """Mark community-scoped ISSUED/SENT invoices past due as OVERDUE. Returns count.""" + async def sweep_overdue(self, today: datetime.date) -> list[SweptInvoice]: + """Mark community-scoped ISSUED/SENT invoices past due as OVERDUE. + + Returns the affected rows rather than a count, because each one has to be + notified to the member it bills. ``synchronize_session=False`` is + required, not stylistic: an explicit ``.returning()`` conflicts with the + default session-synchronisation strategy, and the ORM identity map is not + used on this path. + """ result = await self._session.execute( update(InvoiceModel) .where( @@ -358,8 +379,12 @@ async def sweep_overdue(self, today: datetime.date) -> int: InvoiceModel.due_date < today, ) .values(status=InvoiceStatus.OVERDUE) + .returning(InvoiceModel.id, InvoiceModel.id_member, InvoiceModel.number) + .execution_options(synchronize_session=False) ) - return int(result.rowcount) + return [ + SweptInvoice(id=row.id, id_member=row.id_member, number=row.number) for row in result + ] # ---- payments ---------------------------------------------------------- async def add_payment(self, payment: PaymentModel) -> PaymentModel: diff --git a/api/billing/service.py b/api/billing/service.py index 074011f..bdeab19 100644 --- a/api/billing/service.py +++ b/api/billing/service.py @@ -34,6 +34,13 @@ current_user_role, ) from core.errors.errors import ErrorException +from core.notifications import ( + Channel, + NotificationCategory, + NotificationService, + NotificationTypes, + UsersTarget, +) from core.queue.helper import Event from core.security.user_context import ROLE_HIERARCHY, Role from core.storage import client as storage @@ -88,6 +95,7 @@ def __init__( self._publisher = publisher self._email = email self._audit = audit + self._notify = NotificationService(crm_session) self._settings = settings def _community(self) -> int: @@ -361,6 +369,27 @@ async def issue_invoice(self, *, invoice_id: int) -> IssueOut: ), id_community=cid, ) + # Same CRM transaction as the audit row: both land or neither does. + # `data` holds JSON primitives only — it goes into a JSONB column, so the + # Decimal total and the date have to be strings. + by_member = await self._crm_read.user_ids_for_members( + id_community=cid, member_ids=[invoice.id_member] + ) + await self._notify.publish( + type=NotificationTypes.INVOICE_ISSUED, + target=UsersTarget(user_ids=by_member.get(invoice.id_member, []), community_id=cid), + category=NotificationCategory.TRANSACTIONAL, + channels=(Channel.INAPP, Channel.EMAIL), + data={ + "invoice_id": invoice_id, + # Read from the locals, not from `invoice`: mark_issued is a bare + # UPDATE, so the loaded instance still says DRAFT with no number. + "number": number, + "due_date": due_date.isoformat(), + "total": str(invoice.total), + "currency": invoice.currency, + }, + ) await self._crm.commit() app_metrics.invoices_issued.add(1) @@ -479,11 +508,39 @@ async def list_payments(self, *, invoice_id: int) -> list[PaymentOut]: return [mappers.payment_to_out(payment) for payment in payments] async def sweep_overdue(self) -> OverdueSweepOut: - self._community() + cid = self._community() today = datetime.datetime.now(_SETTLEMENT_TZ).date() - marked = await self._repo.sweep_overdue(today) + # The UPDATE is staged but NOT committed yet, and the ordering is + # load-bearing. `sweep_overdue` is a single UPDATE ... RETURNING over + # ISSUED/SENT rows, and nothing ever moves an invoice back — so once the + # rows say OVERDUE, no later sweep will ever match them again. Committing + # locally first and then swallowing a CRM failure (as this used to do) + # therefore loses the notification and its email PERMANENTLY, with + # nothing left to re-emit them. + # + # Committing the CRM side first inverts the failure: a crash between the + # two commits leaves the invoices still SENT, the next sweep re-flips and + # re-publishes, and `outbound_message.dedupe_key` collapses the duplicate + # email at the cost of one duplicate in-app row. That is exactly the + # at-least-once trade the dedupe key exists to make. + swept = await self._repo.sweep_overdue(today) + if swept: + by_member = await self._crm_read.user_ids_for_members( + id_community=cid, member_ids=[row.id_member for row in swept] + ) + for row in swept: + await self._notify.publish( + type=NotificationTypes.INVOICE_OVERDUE, + target=UsersTarget(user_ids=by_member.get(row.id_member, []), community_id=cid), + category=NotificationCategory.TRANSACTIONAL, + channels=(Channel.INAPP, Channel.EMAIL), + data={"invoice_id": row.id, "number": row.number}, + ) + # Deliberately unguarded: a CRM failure here must 500 and roll the + # local UPDATE back with it, so the sweep can be retried whole. + await self._crm.commit() await self._local.commit() - return OverdueSweepOut(marked=marked) + return OverdueSweepOut(marked=len(swept)) async def create_credit_note(self, *, invoice_id: int, body: CreditNoteIn) -> InvoiceOut: cid = self._community() diff --git a/core/config.py b/core/config.py index 5161139..1e6c343 100644 --- a/core/config.py +++ b/core/config.py @@ -96,6 +96,12 @@ class Settings(BaseSettings): # Watermarked proforma variant rendered for DRAFT invoices (no legal number). INVOICE_PROFORMA_TEMPLATE_URI: str = "s3://optimce-templates/billing/invoice_proforma/v1/" + # ---- Overdue sweep ---- + # Driven from the worker on a daily tick rather than by a caller. The + # `POST /billing-runs/overdue-sweep` route stays, for a manual run. + OVERDUE_SWEEP_ENABLED: bool = True + OVERDUE_SWEEP_HOUR_LOCAL: int = 6 + # ---- Localization ---- DEFAULT_LOCALE: str = "fr-BE" diff --git a/core/notifications/__init__.py b/core/notifications/__init__.py new file mode 100644 index 0000000..9b2dcdc --- /dev/null +++ b/core/notifications/__init__.py @@ -0,0 +1,26 @@ +from core.notifications.contract import ( + MANAGER_ROLES, + Channel, + CommunityTarget, + NotificationCategory, + NotificationTarget, + UsersTarget, + UserTarget, +) +from core.notifications.repository import NotificationRepository +from core.notifications.service import EmailRecipient, NotificationService +from core.notifications.types import NotificationTypes + +__all__ = [ + "MANAGER_ROLES", + "Channel", + "CommunityTarget", + "EmailRecipient", + "NotificationCategory", + "NotificationRepository", + "NotificationService", + "NotificationTarget", + "NotificationTypes", + "UserTarget", + "UsersTarget", +] diff --git a/core/notifications/contract.py b/core/notifications/contract.py new file mode 100644 index 0000000..41ab005 --- /dev/null +++ b/core/notifications/contract.py @@ -0,0 +1,94 @@ +"""Producer-facing notification contract (IMPLEMENTATION_PLAN.md §1.3). + +This module deliberately imports nothing from the service around it, so it is +byte-identical in every producer and lifts verbatim into the standalone +notification service in Phase 2. It mirrors crm-backend's +``src/modules/notifications/api/notification.dtos.ts`` — keep the two in step. +""" + +from collections.abc import Sequence +from dataclasses import dataclass +from enum import IntEnum +from typing import Final, Literal + + +class Channel(IntEnum): + """A delivery channel a producer *requests*. Never guaranteed. + + The values are the on-disk encoding: they land verbatim in the ``SMALLINT`` + ``channel`` columns of ``notification_preference`` and ``outbound_message`` + (Phase 1 step 3), and in the Phase 2 wire payload. Never renumber. + """ + + INAPP = 1 + EMAIL = 2 + + +class NotificationCategory(IntEnum): + """The producer's statement of *kind*, which decides opt-out policy. + + ``TRANSACTIONAL`` — an invoice, an invitation, a missed regulatory deadline. + Consequential, so it overrides the recipient's channel preferences and must + never offer an unsubscribe link. + ``INFORMATIONAL`` — news, digests, reminders. An opt-out must exist and be + honoured. + + Orthogonal to ``channels`` on purpose: the producer states intent, the + notification layer owns policy. That split is the whole reason the layer can + be extracted later, so it holds from day one. + """ + + TRANSACTIONAL = 1 + INFORMATIONAL = 2 + + +# crm-backend owns this vocabulary: +# community_user.role VARCHAR(50) CHECK (role IN ('ADMIN','MANAGER','MEMBER')) +# It is a CRM schema domain rather than an annexe auth enum, which is why these +# are literals and this module stays import-free. ADMIN is included because it +# outranks MANAGER. +MANAGER_ROLES: Final[tuple[str, ...]] = ("ADMIN", "MANAGER") + + +@dataclass(frozen=True, slots=True) +class UserTarget: + """One recipient, by INTERNAL ``app_user.id``.""" + + user_id: int + community_id: int | None = None + kind: Literal["user"] = "user" + + +@dataclass(frozen=True, slots=True) +class UsersTarget: + """An explicit set of recipients, by INTERNAL ``app_user.id``. + + De-duplicated by ``publish``; caller order is preserved so a fan-out stays + deterministic and testable. + """ + + user_ids: Sequence[int] + community_id: int | None = None + kind: Literal["users"] = "users" + + +@dataclass(frozen=True, slots=True) +class CommunityTarget: + """Every member of a community, optionally narrowed and minus one author. + + ``roles`` narrows to ``community_user.role`` values (see ``MANAGER_ROLES``). + + ``exclude_auth_user_id`` is a Keycloak ``sub``, not an internal id: every + Python producer has the sub in ``current_user_id`` and none of them has the + internal id without a CRM read, which ``publish`` already knows how to do. + The TypeScript union has no equivalent because its callers resolve the + audience up front; the field is additive, so the contracts stay compatible. + """ + + community_id: int + roles: Sequence[str] | None = None + exclude_auth_user_id: str | None = None + kind: Literal["community"] = "community" + + +NotificationTarget = UserTarget | UsersTarget | CommunityTarget diff --git a/core/notifications/dedupe.py b/core/notifications/dedupe.py new file mode 100644 index 0000000..39465de --- /dev/null +++ b/core/notifications/dedupe.py @@ -0,0 +1,89 @@ +"""The idempotency key for one queued outbound message. + +Byte-identical across producers, like the rest of this package, and mirrored in +crm-backend's ``src/modules/notifications/shared/notification.dedupe.ts``. Keep +the two in step. +""" + +import hashlib +import json +from typing import Any + +# Hex characters kept from the payload digest. 128 bits. +_PAYLOAD_HASH_LENGTH = 32 +# Hex characters kept from the address digest. +_ADDRESS_HASH_LENGTH = 16 +# ``outbound_message.dedupe_key`` is VARCHAR(200). +_MAX_KEY_LENGTH = 200 + + +def canonical_json(data: dict[str, Any]) -> str: + """Deterministic JSON: keys sorted recursively, no whitespace. + + ``ensure_ascii=False`` so the output matches ``JSON.stringify`` on the + TypeScript side; nothing requires the two languages to agree on a concrete + key (each notification type has exactly one producer), but a gratuitous + divergence is a trap for anyone comparing them. + + Raises on ``Decimal`` / ``date``, deliberately: ``data`` lands in a JSONB + column and callers are required to stringify. That failure is the same one + ``notification.data`` already has, just reached slightly earlier. + """ + return json.dumps(data, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + + +def build_dedupe_key( + *, + channel: int, + type: str, + data: dict[str, Any], + id_user: int | None = None, + recipient: str | None = None, + override: str | None = None, +) -> str: + """Derive the key that makes a redelivery, a re-run sweep and a retry one row. + + ``::u:`` account-ful + ``::a:`` account-less + ``h = sha256(canonical_json(data))[:32]`` + + The channel prefix is required because the table's grain is + (message, channel, recipient), so one notification delivered over two + channels is two rows. The ``u``/``a`` namespaces stop a user id and an + address from ever colliding. + + **``data`` IS the idempotency key, for all time.** There is no time bucket, + so two genuinely distinct occurrences of the same type to the same recipient + with identical ``data`` collapse into a single message, permanently. That is + correct for everything driven by a status transition — ``invoice.issued``, + ``invoice.overdue``, ``admin_deadline.missed``, an invitation — each of which + fires once per row and carries that row's id in ``data``. A producer whose + sweep can re-emit WITHOUT mutating its source row must pass ``override`` + including the occurrence date; ``admin_deadline.due_soon`` is the one that + does. + + Worst case length is 128 (type) + 2 + 10 + 1 + 32 = 175, inside VARCHAR(200). + """ + if override: + return override[:_MAX_KEY_LENGTH] + payload_hash = hashlib.sha256(canonical_json(data).encode("utf-8")).hexdigest()[ + :_PAYLOAD_HASH_LENGTH + ] + if id_user is not None: + recipient_ref = f"u{id_user}" + else: + address = (recipient or "").strip().lower() + digest = hashlib.sha256(address.encode("utf-8")).hexdigest()[:_ADDRESS_HASH_LENGTH] + recipient_ref = f"a{digest}" + return f"{channel}:{type}:{recipient_ref}:{payload_hash}" + + +def type_prefix_of(type: str) -> str: + """The ``notification_preference.type_prefix`` a type falls under. + + Its first dot-segment. ``''`` (the default row) is never produced here. The + taxonomy guarantees exactly two segments, so this is total; a malformed key + degrades to the whole string, which simply matches no preference row. + """ + head, _, _ = type.partition(".") + return head diff --git a/core/notifications/repository.py b/core/notifications/repository.py new file mode 100644 index 0000000..3c296fb --- /dev/null +++ b/core/notifications/repository.py @@ -0,0 +1,155 @@ +from collections.abc import Sequence +from dataclasses import dataclass + +from sqlalchemy import select +from sqlalchemy.dialects.postgresql import insert as pg_insert +from sqlalchemy.ext.asyncio import AsyncSession + +from shared.models.crm_models import AppUserModel, CommunityUserModel, NotificationModel +from shared.models.crm_notification_models import ( + NotificationPreferenceModel, + OutboundMessageModel, +) + + +@dataclass(frozen=True, slots=True) +class RecipientContact: + """Everything the delivery layer needs to address one recipient. + + Read once at enqueue and copied onto the queued row, so a later profile + change never redirects or relabels an already-queued message. + """ + + id_user: int + email: str + locale: str | None + display_name: str | None + + +class NotificationRepository: + """Read community membership / write notifications against the CRM database. + + All the tables here (``app_user``, ``community_user``, ``notification``, + ``outbound_message``, ``notification_preference``) are owned by + ``crm-backend``; an annexe reads the membership and preference sides and + inserts into the notification and outbound sides when one of its domain + events should reach a user. + + Byte-identical across producers — see ``service.py``'s module docstring. + """ + + def __init__(self, session: AsyncSession): + self.session = session + + async def resolve_internal_user_id(self, auth_user_id: str) -> int | None: + """Map a Keycloak ``sub`` to its internal ``app_user.id`` (``None`` if absent).""" + stmt = select(AppUserModel.id).where(AppUserModel.auth_user_id == auth_user_id) + result = await self.session.execute(stmt) + return result.scalar_one_or_none() + + async def find_community_recipient_ids( + self, + community_id: int, + *, + exclude_user_id: int | None = None, + roles: Sequence[str] | None = None, + ) -> list[int]: + """Return the internal ids of a community's members. + + ``exclude_user_id`` drops one member (typically the author) from the + fan-out; ``roles`` optionally narrows to specific roles (unused today, + kept to mirror crm-backend's ``findCommunityRecipientIds``). + """ + stmt = select(CommunityUserModel.id_user).where( + CommunityUserModel.id_community == community_id + ) + if exclude_user_id is not None: + stmt = stmt.where(CommunityUserModel.id_user != exclude_user_id) + if roles: + stmt = stmt.where(CommunityUserModel.role.in_(roles)) + result = await self.session.execute(stmt) + return list(result.scalars().all()) + + async def insert_many(self, rows: list[NotificationModel]) -> None: + """Stage a batch of notification rows. The caller owns the commit. + + Flushes, so the generated ids exist by the time ``_enqueue_email`` needs + them to link each queued message back to its in-app row. + """ + if not rows: + return + self.session.add_all(rows) + await self.session.flush() + + async def find_preferences( + self, user_ids: Sequence[int], type_prefix: str + ) -> dict[tuple[int, int], int]: + """Effective ``mode`` per (user, channel) for one type, most-specific-wins. + + A row whose ``type_prefix`` is the type's first dot-segment beats the + ``''`` default row. Pairs absent from the result have expressed no + preference and default to IMMEDIATE. + + One round trip for the whole audience, and only ever called for + INFORMATIONAL notifications — TRANSACTIONAL bypasses preference entirely + and must not reach this query. + """ + if not user_ids: + return {} + stmt = select( + NotificationPreferenceModel.id_user, + NotificationPreferenceModel.channel, + NotificationPreferenceModel.mode, + NotificationPreferenceModel.type_prefix, + ).where( + NotificationPreferenceModel.id_user.in_(list(user_ids)), + NotificationPreferenceModel.type_prefix.in_(["", type_prefix]), + ) + result = await self.session.execute(stmt) + resolved: dict[tuple[int, int], int] = {} + for id_user, channel, mode, row_prefix in result.all(): + key = (id_user, channel) + # A specific prefix always wins; between two rows of the same + # specificity there can only be one, since (id_user, type_prefix, + # channel) is the primary key. + if row_prefix != "" or key not in resolved: + resolved[key] = mode + return resolved + + async def find_recipient_contacts(self, user_ids: Sequence[int]) -> dict[int, RecipientContact]: + """Resolve email, locale and display name for a set of internal user ids. + + Users with no row are simply absent — the caller queues nothing for them, + which is not an error: the in-app notification stands on its own. + """ + if not user_ids: + return {} + stmt = select( + AppUserModel.id, + AppUserModel.email, + AppUserModel.locale, + AppUserModel.first_name, + AppUserModel.last_name, + ).where(AppUserModel.id.in_(list(user_ids))) + result = await self.session.execute(stmt) + contacts: dict[int, RecipientContact] = {} + for id_user, email, locale, first_name, last_name in result.all(): + name = " ".join(part for part in (first_name, last_name) if part and part.strip()) + contacts[id_user] = RecipientContact( + id_user=id_user, + email=email, + locale=locale, + display_name=name or None, + ) + return contacts + + async def insert_outbound(self, rows: list[dict[str, object]]) -> None: + """Stage outbound rows, skipping any whose ``dedupe_key`` already exists. + + Targeted ``ON CONFLICT``: an untargeted ``DO NOTHING`` would also swallow + a violation of any future unique index on this table. + """ + if not rows: + return + stmt = pg_insert(OutboundMessageModel).values(rows) + await self.session.execute(stmt.on_conflict_do_nothing(index_elements=["dedupe_key"])) diff --git a/core/notifications/service.py b/core/notifications/service.py new file mode 100644 index 0000000..bcb08cc --- /dev/null +++ b/core/notifications/service.py @@ -0,0 +1,285 @@ +"""Publish durable notifications into the shared CRM ``notification`` table. + +Mirrors ``AuditLogService``: the write rides on the caller's CRM session inside a +SAVEPOINT and never raises — a notification failure must not abort the business +write that triggered it. The caller owns the commit. + +This file is byte-identical across news-board, billing and +administrative-document. The only per-service module in this package is +``types.py``. Keep it that way: it is what makes Phase 2's extraction a move +rather than a rewrite. +""" + +import logging +from collections.abc import Sequence +from dataclasses import dataclass +from typing import Any + +from sqlalchemy.ext.asyncio import AsyncSession + +from core.notifications.contract import ( + Channel, + CommunityTarget, + NotificationCategory, + NotificationTarget, + UsersTarget, + UserTarget, +) +from core.notifications.dedupe import build_dedupe_key, type_prefix_of +from core.notifications.repository import NotificationRepository +from shared.models.crm_models import NotificationModel + +logger = logging.getLogger(__name__) + +# `notification_preference.mode`: 1 IMMEDIATE, 3 OFF. Only OFF changes anything, +# so it is the only value this module needs to know. 2 DAILY_DIGEST is reserved +# in the encoding and rejected by a DB CHECK until a digest runner exists. +_PREFERENCE_MODE_OFF = 3 + + +@dataclass(frozen=True, slots=True) +class EmailRecipient: + """One resolved addressee of the EMAIL channel. + + ``id_notification`` is ``None`` when INAPP was not among the effective + channels — exactly the case step 3's ``outbound_message.id_notification + BIGINT NULL`` exists for. + """ + + id_user: int + id_notification: int | None + + +class NotificationService: + """The producer-facing notification API (IMPLEMENTATION_PLAN.md §1.3).""" + + def __init__(self, crm_session: AsyncSession): + self.crm_session = crm_session + self.repository = NotificationRepository(crm_session) + + async def publish( + self, + *, + type: str, + target: NotificationTarget, + category: NotificationCategory, + channels: Sequence[Channel], + data: dict[str, Any] | None = None, + dedupe_key: str | None = None, + ) -> int: + """Fan a notification out to ``target``. Returns the in-app rows staged. + + ``category`` and ``channels`` are both required and both meaningful: + ``channels`` is what the producer *asks* for, ``category`` is what the + producer *is*, and effective delivery is the intersection with the + recipient's preferences — except for TRANSACTIONAL, which overrides + them. Neither is defaulted: a defaulted ``channels`` fails silently (a + producer that meant EMAIL quietly gets INAPP only), while a missing one + fails loudly at the call site. + + ``data`` must hold JSON primitives only — it lands in a JSONB column, so + ``Decimal`` and ``date`` have to be stringified by the caller. It is also + the email idempotency key: see ``dedupe.build_dedupe_key``. + + ``dedupe_key`` overrides that derivation. Set it ONLY when a recurring + sweep can re-emit the same payload for a genuinely new occurrence, and + include the occurrence date. + + Returns 0 on an empty audience, on any swallowed failure, when INAPP is + not among the effective channels (an email-only publish), and when every + recipient has muted the in-app channel. The count is about in-app rows, + not about delivery. + """ + try: + # `no_autoflush` because these reads happen OUTSIDE the savepoint + # below. Without it, SQLAlchemy would flush whatever the caller + # staged earlier (in billing's issue_invoice, the audit row) as a + # side effect of our SELECT — and a failure there would abort the + # caller's CRM transaction outside the savepoint's protection, get + # swallowed by the blanket `except`, and resurface opaquely at their + # commit. That is exactly the class of bug the savepoint exists to + # prevent. The notification layer must never flush a caller's work. + with self.crm_session.no_autoflush: + recipient_ids, community_id = await self._resolve_audience(target) + if not recipient_ids: + return 0 + + effective = await self._effective_channels( + type=type, + category=category, + requested=channels, + recipient_ids=recipient_ids, + ) + + none: frozenset[Channel] = frozenset() + inapp_ids = [uid for uid in recipient_ids if Channel.INAPP in effective.get(uid, none)] + email_ids = [uid for uid in recipient_ids if Channel.EMAIL in effective.get(uid, none)] + if not inapp_ids and not email_ids: + return 0 + + payload = data or {} + rows = [ + NotificationModel( + id_community=community_id, + id_user=user_id, + type=type, + data=payload, + ) + for user_id in inapp_ids + ] + + # ONE savepoint around every write this publish performs. The in-app + # rows and the outbound_message rows must land or vanish together, + # and a failure must leave the caller's CRM transaction clean and + # committable. insert_many flushes, so the notification ids exist by + # the time _enqueue_email needs them. + async with self.crm_session.begin_nested(): + await self.repository.insert_many(rows) + if email_ids: + notification_ids = {row.id_user: row.id for row in rows} + await self._enqueue_email( + type=type, + data=payload, + category=category, + id_community=community_id, + dedupe_key=dedupe_key, + recipients=[ + EmailRecipient( + id_user=user_id, + id_notification=notification_ids.get(user_id), + ) + for user_id in email_ids + ], + ) + return len(rows) + except Exception: + logger.exception( + "notification.publish failed", + extra={"operation": "notification:publish", "type": type}, + ) + return 0 + + async def _resolve_audience(self, target: NotificationTarget) -> tuple[list[int], int | None]: + """Turn a target into (de-duplicated recipient ids, source community).""" + match target: + case UserTarget(): + return [target.user_id], target.community_id + case UsersTarget(): + # dict.fromkeys de-duplicates while preserving caller order, + # mirroring TypeScript's [...new Set(userIds)]. + return list(dict.fromkeys(target.user_ids)), target.community_id + case CommunityTarget(): + exclude_user_id = ( + await self.repository.resolve_internal_user_id(target.exclude_auth_user_id) + if target.exclude_auth_user_id + else None + ) + recipients = await self.repository.find_community_recipient_ids( + target.community_id, + exclude_user_id=exclude_user_id, + roles=target.roles, + ) + return recipients, target.community_id + case _: # pragma: no cover — the union is closed + raise TypeError(f"unsupported notification target: {target!r}") + + async def _effective_channels( + self, + *, + type: str, + category: NotificationCategory, + requested: Sequence[Channel], + recipient_ids: Sequence[int], + ) -> dict[int, frozenset[Channel]]: + """Requested channels ∩ each recipient's preference; TRANSACTIONAL overrides. + + Per-recipient, not per-publish: ``notification_preference`` is keyed by + user and a community fan-out reaches many of them, so a single answer for + everyone would let one manager who muted a reminder mute it for the whole + community. + + ``category is TRANSACTIONAL`` skips the lookup entirely — an invoice or a + missed regulatory deadline is not opt-out-able, so there is nothing to + read and no query to pay for. + """ + default = frozenset(requested) + effective = {user_id: default for user_id in recipient_ids} + if category is NotificationCategory.TRANSACTIONAL: + return effective + + modes = await self.repository.find_preferences(recipient_ids, type_prefix_of(type)) + for (user_id, channel), mode in modes.items(): + if mode != _PREFERENCE_MODE_OFF or user_id not in effective: + continue + effective[user_id] = effective[user_id] - {Channel(channel)} + return effective + + async def _enqueue_email( + self, + *, + type: str, + data: dict[str, Any], + category: NotificationCategory, + id_community: int | None, + recipients: Sequence[EmailRecipient], + dedupe_key: str | None = None, + ) -> None: + """Stage one ``outbound_message`` row per addressee. + + Called on the real path from inside ``publish``'s SAVEPOINT with the + notification ids already flushed, so the queued mail shares the + producer's transaction: the business write committing is what makes the + message queued, and rolling back un-queues it. + + The recipient's address, display name and locale are resolved HERE and + copied onto the row, so a later profile change never redirects an + already-queued message. A recipient with no ``app_user`` row is skipped — + not an error, their in-app notification stands on its own. + + There is deliberately no ``email_suppression`` check at enqueue: a bounce + can land after a message is queued, so only the dispatcher's check before + each send can be authoritative, and doing it twice would mean two places + that must agree on address normalisation. + """ + contacts = await self.repository.find_recipient_contacts( + [recipient.id_user for recipient in recipients] + ) + rows: list[dict[str, object]] = [] + for recipient in recipients: + contact = contacts.get(recipient.id_user) + if contact is None: + continue + address = contact.email.strip() + # A newline in an address splits the SMTP header block, so a + # producer-controlled value could inject headers or extra + # recipients. The queue must never contain one. + if not address or "\r" in address or "\n" in address: + logger.warning( + "notification: rejected an unusable outbound recipient address", + extra={"operation": "notification:enqueue_email", "type": type}, + ) + continue + rows.append( + { + "id_notification": recipient.id_notification, + "id_community": id_community, + "channel": int(Channel.EMAIL), + "recipient": address, + "recipient_name": (contact.display_name or None), + # '' means "unknown": the dispatcher owns the fallback chain, + # because it is the only component that knows which locales + # it actually has templates for. + "locale": contact.locale or "", + "type": type, + "category": int(category), + "data": data, + "dedupe_key": build_dedupe_key( + channel=int(Channel.EMAIL), + type=type, + data=data, + id_user=recipient.id_user, + override=dedupe_key, + ), + } + ) + await self.repository.insert_outbound(rows) diff --git a/core/notifications/types.py b/core/notifications/types.py new file mode 100644 index 0000000..533e783 --- /dev/null +++ b/core/notifications/types.py @@ -0,0 +1,20 @@ +class NotificationTypes: + """Notification ``type`` keys this service publishes. + + ``.`` strings, matching crm-backend's free-form taxonomy. + The frontend localises the displayed text from the key + (``NOTIFICATIONS.TYPES..title``); the backend stores only key + data. + + Every key added here also needs an entry in + ``crm-frontend/src/app/features/notifications/services/notification-type.registry.ts`` + and a title in all four ``crm-frontend/src/assets/i18n/*.json`` files, or it + renders to the user as a raw i18n key with no error anywhere. + + This is the ONLY per-service module in ``core/notifications``; the other four + are byte-identical across producers so Phase 2's extraction is a move rather + than a rewrite. + """ + + INVOICE_ISSUED = "invoice.issued" + INVOICE_OVERDUE = "invoice.overdue" + BILLING_RUN_COMPLETED = "billing_run.completed" diff --git a/ports/crm_core.py b/ports/crm_core.py index 9a7141d..93c953d 100644 --- a/ports/crm_core.py +++ b/ports/crm_core.py @@ -118,3 +118,7 @@ async def participant_contacts( ) -> dict[int, ParticipantContact]: ... async def member_ids_for_user(self, *, id_community: int, auth_user_id: str) -> list[int]: ... + + async def user_ids_for_members( + self, *, id_community: int, member_ids: Sequence[int] + ) -> dict[int, list[int]]: ... diff --git a/ports/crm_core_sqlalchemy.py b/ports/crm_core_sqlalchemy.py index f70db02..397556b 100644 --- a/ports/crm_core_sqlalchemy.py +++ b/ports/crm_core_sqlalchemy.py @@ -288,3 +288,35 @@ async def member_ids_for_user(self, *, id_community: int, auth_user_id: str) -> {"cid": id_community, "auth_user_id": auth_user_id}, ) return [int(row["id"]) for row in result.mappings()] + + async def user_ids_for_members( + self, *, id_community: int, member_ids: Sequence[int] + ) -> dict[int, list[int]]: + """The portal user(s) each member is represented by, keyed by member id. + + The reverse of ``member_ids_for_user``, used to address an invoice + notification to the member it bills. Batched because the overdue sweep + notifies every invoice it flips in one pass. + + A member with no linked account (a company invoiced on paper) is simply + absent from the result. That is not an error — the caller notifies + nobody and carries on. + """ + if not member_ids: + return {} + result = await self._session.execute( + text( + """ + SELECT DISTINCT m.id AS id_member, uml.id_user AS id_user + FROM member m + JOIN user_member_link uml ON uml.id_member = m.id + WHERE m.id_community = :cid AND m.id IN :ids + ORDER BY m.id, uml.id_user + """ + ).bindparams(bindparam("ids", expanding=True)), + {"cid": id_community, "ids": list(member_ids)}, + ) + by_member: dict[int, list[int]] = {} + for row in result.mappings(): + by_member.setdefault(int(row["id_member"]), []).append(int(row["id_user"])) + return by_member diff --git a/ports/providers.py b/ports/providers.py new file mode 100644 index 0000000..65b00ea --- /dev/null +++ b/ports/providers.py @@ -0,0 +1,27 @@ +"""Which adapter backs each port — chosen here, framework-free. + +The choice used to live in ``api/billing/deps.py``, which meant importing it +dragged in ``fastapi``. That is fine for the API and fatal for the worker: +``Dockerfile.worker`` installs ``requirements/worker.txt`` (no fastapi, no +uvicorn) precisely to keep the HTTP stack out of that image, so the scheduler's +``from api.billing.deps import ...`` crash-looped the container on every start. + +So the selection lives here, next to the adapters, and both callers import it: +``api/billing/deps.py`` wraps these in ``Depends``, ``worker/sweeps.py`` calls +them directly. One definition, so the request path and the daily tick cannot +drift onto different adapters. +""" + +from __future__ import annotations + +from ports.email import EmailPort +from ports.email_noop import NoopEmailAdapter +from ports.events import EventPublisher, NatsEventPublisher + + +def get_event_publisher() -> EventPublisher: + return NatsEventPublisher() + + +def get_email_port() -> EmailPort: + return NoopEmailAdapter() diff --git a/shared/models/crm_models.py b/shared/models/crm_models.py index 9f3b49e..55cff62 100644 --- a/shared/models/crm_models.py +++ b/shared/models/crm_models.py @@ -1,13 +1,62 @@ -from sqlalchemy import Integer, String +import datetime +from typing import Any + +from sqlalchemy import TIMESTAMP, BigInteger, Integer, String, Text, func +from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.orm import Mapped, mapped_column from core.database.database import CrmBase class AppUserModel(CrmBase): - # Partial mapping of the CRM `app_user` table: only the columns the audit - # log service needs to denormalise the writer's identity onto each row. + # Partial mapping of the CRM `app_user` table: the columns the audit log + # service needs to denormalise the writer's identity onto each row, plus the + # locale and name pair `core/notifications` reads when it addresses a + # queued email. __tablename__ = "app_user" id: Mapped[int] = mapped_column(Integer, primary_key=True) auth_user_id: Mapped[str] = mapped_column(String(255), nullable=False, unique=True) email: Mapped[str] = mapped_column(String(256), nullable=False) + # Preferred language. NULL for every account created before the column + # existed, which the dispatcher's locale fallback is what handles. + locale: Mapped[str | None] = mapped_column(String(8), nullable=True) + first_name: Mapped[str | None] = mapped_column(Text, nullable=True) + last_name: Mapped[str | None] = mapped_column(Text, nullable=True) + + +class CommunityUserModel(CrmBase): + """Partial mapping of the CRM ``community_user`` join table. + + The membership roster of a community (one row per user, with their role). + Read to narrow a notification fan-out to a community's managers. Owned by + ``crm-backend``; read-only here. + """ + + __tablename__ = "community_user" + id_community: Mapped[int] = mapped_column(Integer, primary_key=True) + id_user: Mapped[int] = mapped_column(Integer, primary_key=True) + role: Mapped[str] = mapped_column(String(50), nullable=False) + + +class NotificationModel(CrmBase): + """Mapping of the shared CRM ``notification`` table. + + A durable, per-recipient notification row (one row per user). The table is + owned by ``crm-backend`` — which serves the read API the frontend polls — so + this service only ever *inserts* here, through ``core/notifications``; + ``read_at``/``created_at`` and the bigint ``id`` are managed by the DB. + Mirrors the ``AuditLogModel`` conventions in ``core/database/models.py``. + """ + + __tablename__ = "notification" + id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + id_community: Mapped[int | None] = mapped_column(Integer, nullable=True) + id_user: Mapped[int] = mapped_column(Integer, nullable=False) + type: Mapped[str] = mapped_column(String(128), nullable=False) + data: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False, default=dict) + read_at: Mapped[datetime.datetime | None] = mapped_column( + TIMESTAMP(timezone=True), nullable=True + ) + created_at: Mapped[datetime.datetime] = mapped_column( + TIMESTAMP(timezone=True), nullable=False, server_default=func.now() + ) diff --git a/shared/models/crm_notification_models.py b/shared/models/crm_notification_models.py new file mode 100644 index 0000000..7d102d9 --- /dev/null +++ b/shared/models/crm_notification_models.py @@ -0,0 +1,128 @@ +"""Mappings of the CRM notification-delivery tables. + +All three are owned by ``crm-backend`` (see +``database_script/2026-08-03_notification_delivery.sql``). A producer annexe +INSERTs into ``outbound_message`` and SELECTs from ``notification_preference`` +through ``core/notifications``; ``email_suppression`` is mapped for completeness +and is written only by the ``notification-dispatch`` worker. + +**This module is byte-identical across every service that carries it** — +news-board, billing, administrative-document and notification-dispatch — and is +covered by the same diff gate as ``core/notifications/*.py`` +(IMPLEMENTATION_PLAN §1.8). It deliberately imports nothing but ``CrmBase``, so +it lifts out verbatim in Phase 2. It is kept separate from ``crm_models.py`` +precisely because that file is NOT byte-identical (its docstrings differ per +service), and the byte-identical ``repository.py`` binds to these definitions: +a one-character divergence in a column length would be an invisible, +per-service runtime bug. + +The integer codes are the on-disk encoding shared with +``core/notifications/contract.py`` and crm-backend's ``notification.types.ts``. +Never renumber them. +""" + +import datetime +from typing import Any + +from sqlalchemy import TIMESTAMP, BigInteger, Integer, SmallInteger, String, Text, func +from sqlalchemy.dialects.postgresql import JSONB +from sqlalchemy.orm import Mapped, mapped_column + +from core.database.database import CrmBase + + +class OutboundMessageModel(CrmBase): + """One queued outbound message: (message, channel, recipient). + + Staged inside ``publish``'s SAVEPOINT on the producer's own transaction, so + "the business write committed => the message is queued" is an invariant + rather than a hope. Driven to completion by ``notification-dispatch``, which + is the only thing that ever writes a ``status`` past PENDING. + + ``id_notification`` is nullable because an invitation to an address with no + account has no notification row to hang off (``notification.id_user`` is NOT + NULL). ``recipient`` / ``recipient_name`` are resolved at enqueue time and + stored literally, so a later change of address never redirects an + already-queued message — which is also why there is no ``id_user`` column + and no join back to ``app_user``. + """ + + __tablename__ = "outbound_message" + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + id_notification: Mapped[int | None] = mapped_column(BigInteger, nullable=True) + id_community: Mapped[int | None] = mapped_column(Integer, nullable=True) + # Channel: 1 INAPP, 2 EMAIL + channel: Mapped[int] = mapped_column(SmallInteger, nullable=False) + recipient: Mapped[str] = mapped_column(String(320), nullable=False) + recipient_name: Mapped[str | None] = mapped_column(String(255), nullable=True) + # '' means "unknown"; the dispatcher applies its own default locale. + locale: Mapped[str] = mapped_column(String(8), nullable=False, server_default="") + type: Mapped[str] = mapped_column(String(128), nullable=False) + # NotificationCategory: 1 TRANSACTIONAL, 2 INFORMATIONAL. Persisted because + # the dispatcher decides from it whether to render an opt-out footer, and + # deriving that from `type` would duplicate the producer's policy statement. + category: Mapped[int] = mapped_column(SmallInteger, nullable=False) + data: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False, default=dict) + dedupe_key: Mapped[str] = mapped_column(String(200), nullable=False) + # 1 PENDING, 2 SENT, 3 FAILED, 4 SUPPRESSED, 5 CLAIMED + status: Mapped[int] = mapped_column(SmallInteger, nullable=False, server_default="1") + attempts: Mapped[int] = mapped_column(SmallInteger, nullable=False, server_default="0") + last_error: Mapped[str | None] = mapped_column(Text, nullable=True) + scheduled_for: Mapped[datetime.datetime] = mapped_column( + TIMESTAMP(timezone=True), nullable=False, server_default=func.now() + ) + claimed_at: Mapped[datetime.datetime | None] = mapped_column( + TIMESTAMP(timezone=True), nullable=True + ) + sent_at: Mapped[datetime.datetime | None] = mapped_column( + TIMESTAMP(timezone=True), nullable=True + ) + created_at: Mapped[datetime.datetime] = mapped_column( + TIMESTAMP(timezone=True), nullable=False, server_default=func.now() + ) + + +class EmailSuppressionModel(CrmBase): + """An address that must never be emailed again. Stored LOWER-CASED. + + ``app_user.email`` is a case-sensitive TEXT UNIQUE and providers report + bounces with arbitrary case, so every write path normalises or the list + silently misses. + + Written only by ``notification-dispatch``: at enqueue there is nothing useful + to check (a bounce can land after a message is queued), and the worker's + check before each send is the only one that can be authoritative. + """ + + __tablename__ = "email_suppression" + + email: Mapped[str] = mapped_column(String(320), primary_key=True) + # 1 HARD_BOUNCE, 2 COMPLAINT, 3 UNSUBSCRIBED, 4 MANUAL + reason: Mapped[int] = mapped_column(SmallInteger, nullable=False) + detail: Mapped[str | None] = mapped_column(Text, nullable=True) + created_at: Mapped[datetime.datetime] = mapped_column( + TIMESTAMP(timezone=True), nullable=False, server_default=func.now() + ) + + +class NotificationPreferenceModel(CrmBase): + """What a recipient wants done with a (type prefix, channel) pair. + + Read only for INFORMATIONAL notifications: TRANSACTIONAL overrides + preference entirely and never touches this table, because an invoice, an + invitation or a missed regulatory deadline is not opt-out-able. + + ``type_prefix`` is ``''`` for the catch-all default, else the FIRST + dot-segment of a type key (``invoice``, ``admin_deadline``, …). Resolution is + most-specific-wins per (user, channel); an absent row means IMMEDIATE. + """ + + __tablename__ = "notification_preference" + + id_user: Mapped[int] = mapped_column(Integer, primary_key=True) + type_prefix: Mapped[str] = mapped_column(String(128), primary_key=True) + # Channel: 1 INAPP, 2 EMAIL + channel: Mapped[int] = mapped_column(SmallInteger, primary_key=True) + # 1 IMMEDIATE, 3 OFF (2 DAILY_DIGEST reserved; no runner, rejected by a CHECK) + mode: Mapped[int] = mapped_column(SmallInteger, nullable=False) diff --git a/tests/billing/test_invoice_pdf.py b/tests/billing/test_invoice_pdf.py index 57b386d..a699d11 100644 --- a/tests/billing/test_invoice_pdf.py +++ b/tests/billing/test_invoice_pdf.py @@ -458,3 +458,75 @@ async def test_docgen_success_recovers_render_failed(client, db_session): fresh = await db_session.get(InvoiceModel, invoice_id) assert fresh.status == InvoiceStatus.ISSUED # recovered from RENDER_FAILED assert fresh.artifact_uri == _URI + + +# --------------------------------------------------------------------------- +# Wire-contract regression: the docgen result has NO top-level tenant_id. +# +# document-generation's GenerationResult is extra="forbid" and declares exactly +# {request_id, status, artifacts, error, template_version, generated_at, metadata} +# -- tenant_id exists only on the *request*. Every test above hand-builds a body +# with a top-level tenant_id the service never emits, so they all passed while +# production dropped every real result and no invoice PDF ever attached. +# --------------------------------------------------------------------------- + + +def _wire_result(request_id: str, cid: int, **overrides) -> dict: + """A body shaped exactly like ``GenerationResult.to_json_bytes()`` output. + + Note the absence of a top-level ``tenant_id``: the tenant round-trips only + through ``metadata``, which document-generation echoes verbatim. + """ + body = { + "request_id": request_id, + "status": "success", + "artifacts": [{"format": "pdf", "uri": _URI, "size_bytes": 1234, "sha256": "ok"}], + "template_version": "1.0.0", + "generated_at": "2026-07-26T10:00:00Z", + "metadata": {"invoice_id": 1, "tenant_id": str(cid)}, + } + body.update(overrides) + return body + + +async def test_issue_puts_tenant_id_in_metadata(client, db_session): + """The request half of the contract: without this the result cannot be routed.""" + cid, _member, invoice = await _seed_single(client, db_session) + await _issue(client, invoice["id"]) + + fake = _FakeDocGen() + await issue.process_issue( + invoice["id"], doc_port=fake, local_session=db_session, crm_session=db_session + ) + assert fake.requests[0].metadata["tenant_id"] == str(cid) + + +async def test_docgen_result_without_top_level_tenant_id_attaches(client, db_session): + cid, _member, invoice = await _seed_single(client, db_session) + invoice_id = invoice["id"] + await _issue(client, invoice_id) + await _set(db_session, invoice_id, docgen_request_id="wire-req") + + outcome = await docgen_results.process_docgen_result( + _wire_result("wire-req", cid), + local_session=db_session, + crm_session=db_session, + ) + assert outcome == "attached" + + db_session.expire_all() + fresh = await db_session.get(InvoiceModel, invoice_id) + assert fresh.artifact_uri == _URI + + +async def test_docgen_result_with_neither_tenant_source_drops(client, db_session): + _cid, _member, invoice = await _seed_single(client, db_session) + await _issue(client, invoice["id"]) + await _set(db_session, invoice["id"], docgen_request_id="no-tenant-req") + + outcome = await docgen_results.process_docgen_result( + {"request_id": "no-tenant-req", "status": "success", "artifacts": [], "metadata": {}}, + local_session=db_session, + crm_session=db_session, + ) + assert outcome == "drop" diff --git a/tests/billing/test_notifications.py b/tests/billing/test_notifications.py new file mode 100644 index 0000000..a46a666 --- /dev/null +++ b/tests/billing/test_notifications.py @@ -0,0 +1,417 @@ +"""Notifications this service writes into the shared CRM ``notification`` table. + +Three producers (IMPLEMENTATION_PLAN §1.4): ``invoice.issued`` and +``invoice.overdue`` reach the member the invoice bills, ``billing_run.completed`` +reaches the community's managers. All three ride on the caller's CRM session and +are staged before the CRM commit, so they share one transaction with the audit +row. ``get_crm_session`` and ``get_local_session`` both resolve to ``db_session`` +here, so the rows are visible to the test without a commit. +""" + +from __future__ import annotations + +import datetime + +import pytest +from sqlalchemy import select, update + +import main +from api.billing.deps import get_event_publisher +from core.database.models import AuditLogModel +from core.notifications import Channel, NotificationCategory, UsersTarget +from core.notifications.repository import NotificationRepository +from core.notifications.service import EmailRecipient, NotificationService +from shared.models.crm_models import NotificationModel +from shared.models.crm_notification_models import ( + NotificationPreferenceModel, + OutboundMessageModel, +) +from shared.models.local_models import InvoiceModel +from tests.factories import crm_billing_factory as f +from worker import persistence + +_AUTH = "notification-test-org" + + +def _headers() -> dict[str, str]: + return { + "x-user-id": "u1", + "x-community-id": _AUTH, + "x-user-orgs": f"[orgId:{_AUTH} orgPath:/x roles:[ADMIN]]", + } + + +class _FakePublisher: + async def publish(self, subject: str, event) -> None: + return None + + +async def _notifications(db_session) -> list[NotificationModel]: + result = await db_session.execute(select(NotificationModel).order_by(NotificationModel.id)) + return list(result.scalars().all()) + + +async def _outbound(db_session) -> list[OutboundMessageModel]: + result = await db_session.execute( + select(OutboundMessageModel).order_by(OutboundMessageModel.id) + ) + return list(result.scalars().all()) + + +async def _seed_run( + db_session, + client, + *, + link_user: bool = True, + roster: dict[str, str] | None = None, +) -> dict[str, int]: + """A community with one billed member, priced consumption and a COMPUTED run. + + Returns the ids the tests assert on. ``link_user=False`` models a member + invoiced on paper: they exist, but no portal account represents them. + ``roster`` seeds ``community_user`` rows (``{auth_id: role}``) *before* the + run is processed, so the ``billing_run.completed`` fan-out sees them; their + internal ids come back under the same keys. + """ + main.app.dependency_overrides[get_event_publisher] = lambda: _FakePublisher() + cid = await f.create_community( + db_session, auth_community_id=_AUTH, iban="BE68539007547034", legal_name="ACME ASBL" + ) + await f.create_subscription(db_session, id_community=cid, feature="billing", is_active=True) + op = await f.create_sharing_operation(db_session, id_community=cid) + member = await f.create_member(db_session, id_community=cid, name="Alice", member_type=1) + await f.create_individual(db_session, id_member=member, email="alice@example.be") + user_id = await f.create_app_user(db_session, auth_user_id="alice-sub") + if link_user: + await f.link_user_to_member(db_session, id_user=user_id, id_member=member) + await f.create_meter(db_session, ean="EAN-N", id_community=cid) + await f.create_meter_data( + db_session, + ean="EAN-N", + id_community=cid, + id_sharing_operation=op, + id_member=member, + client_type=1, + start_date=datetime.date(2026, 1, 1), + ) + await f.create_meter_consumption( + db_session, + ean="EAN-N", + id_community=cid, + id_sharing_operation=op, + timestamp=f.june(5), + shared=30.0, + ) + await client.post( + f"/sharing-operations/{op}/tariffs", + headers=_headers(), + json={"kind": 1, "scope": 1, "price_per_kwh": "0.15", "valid_from": "2026-01-01"}, + ) + ids: dict[str, int] = {} + for auth_id, role in (roster or {}).items(): + uid = await f.create_app_user(db_session, auth_user_id=auth_id) + await f.add_community_member(db_session, id_community=cid, id_user=uid, role=role) + ids[auth_id] = uid + resp = await client.post( + f"/sharing-operations/{op}/billing-runs", + headers=_headers(), + json={"period_start": "2026-06-01", "period_end": "2026-06-30"}, + ) + run_id = resp.json()["data"]["id"] + await persistence.process_billing_run(run_id, local_session=db_session, crm_session=db_session) + listed = await client.get(f"/billing-runs/{run_id}/invoices", headers=_headers()) + ids.update( + cid=cid, + run_id=run_id, + member=member, + user_id=user_id, + invoice_id=listed.json()["data"][0]["id"], + ) + return ids + + +# ---- invoice.issued -------------------------------------------------------- + + +async def test_issue_invoice_notifies_the_invoiced_member(client, db_session): + ids = await _seed_run(db_session, client) + + resp = await client.post(f"/invoices/{ids['invoice_id']}/issue", headers=_headers()) + assert resp.status_code == 200, resp.text + issued = resp.json()["data"] + + notifs = [n for n in await _notifications(db_session) if n.type == "invoice.issued"] + assert len(notifs) == 1 + assert notifs[0].id_user == ids["user_id"] + assert notifs[0].id_community == ids["cid"] + assert notifs[0].read_at is None + # Compared exactly: `data` is JSONB serialised with plain json.dumps, which + # raises on Decimal and date. That raise happens inside publish's savepoint + # and inside its blanket except, so a non-primitive payload silently drops + # the notification behind a 200. This assertion is the only thing that sees it. + assert notifs[0].data == { + "invoice_id": ids["invoice_id"], + "number": issued["number"], + "due_date": issued["due_date"], + "total": "5.45", # 30 kWh at 0.15 plus VAT, stringified from Decimal + "currency": "EUR", + } + + +async def test_issue_invoice_with_unlinked_member_notifies_nobody(client, db_session): + ids = await _seed_run(db_session, client, link_user=False) + + resp = await client.post(f"/invoices/{ids['invoice_id']}/issue", headers=_headers()) + assert resp.status_code == 200, resp.text + assert await _notifications(db_session) == [] + + +async def test_notification_failure_does_not_abort_issue(client, db_session, monkeypatch): + """The SAVEPOINT, not the swallow, is what keeps the CRM transaction usable. + + The audit row is staged on the same CRM session *before* the notification. If + a failed insert poisoned that transaction, the audit row would vanish too — + which is the specific risk of hooking publish after the audit call. + """ + ids = await _seed_run(db_session, client) + + async def _boom(self: NotificationRepository, rows: list[NotificationModel]) -> None: + raise RuntimeError("notification store unavailable") + + monkeypatch.setattr(NotificationRepository, "insert_many", _boom) + + resp = await client.post(f"/invoices/{ids['invoice_id']}/issue", headers=_headers()) + assert resp.status_code == 200, resp.text + assert resp.json()["data"]["status"] == 1 # ISSUED, with its number claimed + + # The rolled-back savepoint leaves no notification rows ... + assert await _notifications(db_session) == [] + # ... and the sibling audit row, staged on the same CRM transaction *before* + # the notification, survived it. + audit = await db_session.execute( + select(AuditLogModel.action).where(AuditLogModel.entity_id == str(ids["invoice_id"])) + ) + assert "billing.invoice.issued" in list(audit.scalars().all()) + # ... and the invoice really is numbered. + number = await db_session.execute( + select(InvoiceModel.number).where(InvoiceModel.id == ids["invoice_id"]) + ) + assert number.scalar_one() is not None + + +async def test_issue_invoice_queues_the_email(client, db_session): + """The real enqueue, not the spy: EMAIL now produces an `outbound_message`. + + The address, display name and locale are resolved HERE and copied onto the + row, so a later profile change never redirects an already-queued message. + """ + ids = await _seed_run(db_session, client) + + resp = await client.post(f"/invoices/{ids['invoice_id']}/issue", headers=_headers()) + assert resp.status_code == 200, resp.text + + queued = await _outbound(db_session) + assert len(queued) == 1 + row = queued[0] + assert row.type == "invoice.issued" + assert row.channel == int(Channel.EMAIL) + assert row.category == int(NotificationCategory.TRANSACTIONAL) + assert row.id_community == ids["cid"] + assert row.status == 1 # PENDING + assert row.attempts == 0 + # The link back to the in-app row is what `insert_many`'s flush buys. + notifs = [n for n in await _notifications(db_session) if n.type == "invoice.issued"] + assert row.id_notification == notifs[0].id + # No locale is seeded, so '' means "unknown" and the dispatcher's fallback + # chain decides — it is the only component that knows which locales it has + # templates for. + assert row.locale == "" + + +async def test_queued_email_is_idempotent_on_the_dedupe_key(client, db_session): + """A replayed publish collapses to one queued message. + + The likeliest implementation bug is deriving the key from `id_notification`, + which would make every key unique and the whole mechanism a silent no-op. + """ + ids = await _seed_run(db_session, client) + await client.post(f"/invoices/{ids['invoice_id']}/issue", headers=_headers()) + + service = NotificationService(db_session) + payload = {"invoice_id": ids["invoice_id"], "number": "X"} + for _ in range(2): + await service.publish( + type="invoice.issued", + target=UsersTarget(user_ids=[ids["user_id"]], community_id=ids["cid"]), + category=NotificationCategory.TRANSACTIONAL, + channels=(Channel.INAPP, Channel.EMAIL), + data=payload, + ) + + # One from the issue above plus one from the pair of identical publishes. + assert len(await _outbound(db_session)) == 2 + + +async def test_queued_email_is_skipped_for_an_unlinked_member(client, db_session): + """A member invoiced on paper has no account, so there is nothing to email.""" + ids = await _seed_run(db_session, client, link_user=False) + + resp = await client.post(f"/invoices/{ids['invoice_id']}/issue", headers=_headers()) + assert resp.status_code == 200, resp.text + assert await _outbound(db_session) == [] + + +async def test_enqueue_failure_does_not_abort_issue(client, db_session, monkeypatch): + """The savepoint covers the enqueue too, not just the notification insert.""" + ids = await _seed_run(db_session, client) + + async def _boom(self: NotificationRepository, rows: list[dict[str, object]]) -> None: + raise RuntimeError("outbound queue unavailable") + + monkeypatch.setattr(NotificationRepository, "insert_outbound", _boom) + + resp = await client.post(f"/invoices/{ids['invoice_id']}/issue", headers=_headers()) + assert resp.status_code == 200, resp.text + + # The whole savepoint rolled back, so neither half of the publish survived... + assert await _outbound(db_session) == [] + assert [n for n in await _notifications(db_session) if n.type == "invoice.issued"] == [] + # ... and the audit row staged before it on the same CRM transaction did. + audit = await db_session.execute( + select(AuditLogModel.action).where(AuditLogModel.entity_id == str(ids["invoice_id"])) + ) + assert "billing.invoice.issued" in list(audit.scalars().all()) + + +async def test_preference_off_mutes_informational_but_never_transactional(client, db_session): + """The pair is the point. + + A preference that silenced an invoice would be a compliance bug, not a + feature — TRANSACTIONAL skips the preference lookup entirely. + """ + ids = await _seed_run(db_session, client) + db_session.add( + NotificationPreferenceModel( + id_user=ids["user_id"], type_prefix="", channel=int(Channel.EMAIL), mode=3 + ) + ) + await db_session.flush() + + service = NotificationService(db_session) + informational = await service.publish( + type="invoice.issued", + target=UsersTarget(user_ids=[ids["user_id"]], community_id=ids["cid"]), + category=NotificationCategory.INFORMATIONAL, + channels=(Channel.INAPP, Channel.EMAIL), + data={"invoice_id": 1}, + ) + assert informational == 1 + assert await _outbound(db_session) == [] + + transactional = await service.publish( + type="invoice.issued", + target=UsersTarget(user_ids=[ids["user_id"]], community_id=ids["cid"]), + category=NotificationCategory.TRANSACTIONAL, + channels=(Channel.INAPP, Channel.EMAIL), + data={"invoice_id": 2}, + ) + assert transactional == 1 + assert len(await _outbound(db_session)) == 1 + + +async def test_email_channel_reaches_the_enqueue_seam(client, db_session, monkeypatch): + """`_enqueue_email` is reached with the recipients already resolved. + + Grepping for it still finds every place the delivery layer hangs off. The + non-None `id_notification` proves the in-app rows were flushed first, which + is what `outbound_message.id_notification` needs. + """ + ids = await _seed_run(db_session, client) + calls: list[dict[str, object]] = [] + + async def _spy(self: NotificationService, **kwargs: object) -> None: + calls.append(kwargs) + + monkeypatch.setattr(NotificationService, "_enqueue_email", _spy) + + resp = await client.post(f"/invoices/{ids['invoice_id']}/issue", headers=_headers()) + assert resp.status_code == 200, resp.text + + assert len(calls) == 1 + assert calls[0]["type"] == "invoice.issued" + assert calls[0]["category"] is NotificationCategory.TRANSACTIONAL + recipients = calls[0]["recipients"] + assert isinstance(recipients, list) + assert [r.id_user for r in recipients] == [ids["user_id"]] + assert all(isinstance(r, EmailRecipient) and r.id_notification is not None for r in recipients) + + +# ---- invoice.overdue ------------------------------------------------------- + + +async def test_overdue_sweep_notifies_the_invoiced_member(client, db_session): + ids = await _seed_run(db_session, client) + await client.post(f"/invoices/{ids['invoice_id']}/issue", headers=_headers()) + await db_session.execute( + update(InvoiceModel) + .where(InvoiceModel.id == ids["invoice_id"]) + .values(due_date=datetime.date(2020, 1, 1)) + ) + + resp = await client.post("/billing-runs/overdue-sweep", headers=_headers()) + assert resp.status_code == 200, resp.text + assert resp.json()["data"]["marked"] == 1 + + overdue = [n for n in await _notifications(db_session) if n.type == "invoice.overdue"] + assert len(overdue) == 1 + assert overdue[0].id_user == ids["user_id"] + assert overdue[0].data["invoice_id"] == ids["invoice_id"] + + +async def test_second_overdue_sweep_notifies_nobody(client, db_session): + """Idempotency comes free: the rows are already OVERDUE, so nothing matches.""" + ids = await _seed_run(db_session, client) + await client.post(f"/invoices/{ids['invoice_id']}/issue", headers=_headers()) + await db_session.execute( + update(InvoiceModel) + .where(InvoiceModel.id == ids["invoice_id"]) + .values(due_date=datetime.date(2020, 1, 1)) + ) + await client.post("/billing-runs/overdue-sweep", headers=_headers()) + + resp = await client.post("/billing-runs/overdue-sweep", headers=_headers()) + assert resp.json()["data"]["marked"] == 0 + + overdue = [n for n in await _notifications(db_session) if n.type == "invoice.overdue"] + assert len(overdue) == 1 + + +# ---- billing_run.completed ------------------------------------------------- + + +_ROSTER = {"admin-sub": "ADMIN", "manager-sub": "MANAGER", "member-sub": "MEMBER"} + + +async def test_billing_run_completed_notifies_managers_only(client, db_session): + ids = await _seed_run(db_session, client, roster=_ROSTER) + + completed = [n for n in await _notifications(db_session) if n.type == "billing_run.completed"] + assert {n.id_user for n in completed} == {ids["admin-sub"], ids["manager-sub"]} + assert ids["member-sub"] not in {n.id_user for n in completed} + assert all(n.id_community == ids["cid"] for n in completed) + assert all(n.data == {"run_id": ids["run_id"], "invoice_count": 1} for n in completed) + + +async def test_redelivered_billing_run_does_not_re_notify(client, db_session): + """A redelivery finds the run already COMPUTED and returns before notifying.""" + ids = await _seed_run(db_session, client, roster={"manager-sub": "MANAGER"}) + + await persistence.process_billing_run( + ids["run_id"], local_session=db_session, crm_session=db_session + ) + + completed = [n for n in await _notifications(db_session) if n.type == "billing_run.completed"] + assert len(completed) == 1 + + +pytestmark = pytest.mark.asyncio diff --git a/tests/billing/test_scheduler.py b/tests/billing/test_scheduler.py new file mode 100644 index 0000000..9dbd3c1 --- /dev/null +++ b/tests/billing/test_scheduler.py @@ -0,0 +1,147 @@ +"""The overdue scheduler: tenant enumeration, the lock, and the next-run clock. + +`POST /billing-runs/overdue-sweep` existed from the start and nothing ever +invoked it, so no invoice was ever marked overdue on its own and +`invoice.overdue` — TRANSACTIONAL, to the member's inbox — could not fire. +""" + +from __future__ import annotations + +import datetime +from zoneinfo import ZoneInfo + +import pytest +from sqlalchemy import text, update + +from core.context_vars import current_internal_community_id +from shared.models.local_models import InvoiceModel +from tests.billing.test_notifications import _headers, _notifications, _outbound, _seed_run +from worker.scheduler import seconds_until_next_run +from worker.sweeps import ( + _communities_with_overdue_invoices_unscoped, + sweep_overdue_for_every_community, +) + +BRUSSELS = ZoneInfo("Europe/Brussels") + + +class TestNextRunClock: + """Pure, so the wrap-around cases are testable without waiting a day.""" + + def test_a_time_before_the_hour_waits_until_today(self): + now = datetime.datetime(2026, 8, 3, 4, 0, tzinfo=BRUSSELS) + assert seconds_until_next_run(now, 6) == 2 * 3600 + + def test_a_time_after_the_hour_waits_until_tomorrow(self): + now = datetime.datetime(2026, 8, 3, 7, 0, tzinfo=BRUSSELS) + assert seconds_until_next_run(now, 6) == 23 * 3600 + + def test_a_utc_instant_is_converted_before_comparing(self): + """The hour is a LOCAL one; comparing in UTC would drift with DST.""" + now = datetime.datetime(2026, 8, 3, 3, 0, tzinfo=datetime.UTC) + assert seconds_until_next_run(now, 6) == 3600 + + +async def _issue_and_backdate(client, db_session) -> dict[str, int]: + ids = await _seed_run(db_session, client) + await client.post(f"/invoices/{ids['invoice_id']}/issue", headers=_headers()) + await db_session.execute( + update(InvoiceModel) + .where(InvoiceModel.id == ids["invoice_id"]) + .values(due_date=datetime.date(2020, 1, 1)) + ) + await db_session.flush() + return ids + + +async def test_the_enumeration_is_unscoped_and_finds_work_with_no_tenant_set(client, db_session): + """The failure this guards against is silent. + + Every owned-DB read in the service filters on the + `current_internal_community_id` ContextVar, which a scheduler has not set — + and a scoped read returns nothing, with no error, which is + indistinguishable from "no work to do". + """ + ids = await _issue_and_backdate(client, db_session) + token = current_internal_community_id.set(None) + try: + found = await _communities_with_overdue_invoices_unscoped( + db_session, today=datetime.date.today() + ) + finally: + current_internal_community_id.reset(token) + assert ids["cid"] in found + + +async def test_the_scheduled_sweep_marks_and_notifies_like_the_route(client, db_session): + """One implementation, two callers — the point of not forking the sweep.""" + ids = await _issue_and_backdate(client, db_session) + current_internal_community_id.set(None) + + marked, swept = await sweep_overdue_for_every_community( + local_session=db_session, crm_session=db_session + ) + + assert (marked, swept) == (1, 1) + overdue = [n for n in await _notifications(db_session) if n.type == "invoice.overdue"] + assert len(overdue) == 1 + assert overdue[0].id_user == ids["user_id"] + # And the email half is queued, which is the whole reason the sweep matters. + assert [row.type for row in await _outbound(db_session)] == [ + "invoice.issued", + "invoice.overdue", + ] + + +async def test_a_second_scheduled_run_is_a_no_op(client, db_session): + """Re-running on a deploy must not re-notify: the invoices are already + OVERDUE, so the UPDATE matches nothing.""" + await _issue_and_backdate(client, db_session) + current_internal_community_id.set(None) + + await sweep_overdue_for_every_community(local_session=db_session, crm_session=db_session) + before = len(await _notifications(db_session)) + + marked, _ = await sweep_overdue_for_every_community( + local_session=db_session, crm_session=db_session + ) + assert marked == 0 + assert len(await _notifications(db_session)) == before + + +async def test_nothing_pending_means_no_work(db_session): + current_internal_community_id.set(None) + assert await sweep_overdue_for_every_community( + local_session=db_session, crm_session=db_session + ) == (0, 0) + + +async def test_the_advisory_lock_is_visible_as_held(db_session): + """Two replicas both sweeping would give every member two bell entries: + `dedupe_key` collapses the duplicate email, nothing collapses the in-app row.""" + from worker.sweeps import _SWEEP_ADVISORY_LOCK_KEY + + acquired = await db_session.scalar( + text("SELECT pg_try_advisory_lock(:k)"), {"k": _SWEEP_ADVISORY_LOCK_KEY} + ) + assert acquired is True + try: + held = await db_session.scalar( + text( + "SELECT EXISTS (SELECT 1 FROM pg_locks " "WHERE locktype = 'advisory' AND granted)" + ) + ) + assert held is True + finally: + await db_session.execute( + text("SELECT pg_advisory_unlock(:k)"), {"k": _SWEEP_ADVISORY_LOCK_KEY} + ) + + +@pytest.mark.parametrize("other_key", [0x0AD3_0001]) +def test_the_billing_and_deadline_locks_do_not_collide(other_key: int): + """Both workers run against the same cluster; a shared key would serialise + two unrelated sweeps and silently hide one of them.""" + from worker.sweeps import _SWEEP_ADVISORY_LOCK_KEY + + assert other_key != _SWEEP_ADVISORY_LOCK_KEY diff --git a/tests/factories/crm_billing_factory.py b/tests/factories/crm_billing_factory.py index d64e3cd..7357bd8 100644 --- a/tests/factories/crm_billing_factory.py +++ b/tests/factories/crm_billing_factory.py @@ -306,6 +306,18 @@ async def link_user_to_member(session: AsyncSession, *, id_user: int, id_member: ) +async def add_community_member( + session: AsyncSession, *, id_community: int, id_user: int, role: str = "MEMBER" +) -> None: + """A CRM community_user row — the roster a manager-targeted fan-out reads.""" + await session.execute( + text( + "INSERT INTO community_user (id_community, id_user, role) " "VALUES (:cid, :uid, :role)" + ), + {"cid": id_community, "uid": id_user, "role": role}, + ) + + def june(day: int, hour: int = 12) -> datetime.datetime: """A tz-aware June 2026 instant (Brussels), for consumption timestamps.""" from zoneinfo import ZoneInfo diff --git a/tests/sql/crm_test_schema.sql b/tests/sql/crm_test_schema.sql index 845adc6..10b7604 100644 --- a/tests/sql/crm_test_schema.sql +++ b/tests/sql/crm_test_schema.sql @@ -1,5 +1,6 @@ --- Test-only DDL for the CRM tables this service READS (and the audit_log it --- writes). The real CRM schema is owned by crm-backend; we mirror only the +-- Test-only DDL for the CRM tables this service READS (and the audit_log and +-- notification rows it writes). The real CRM schema is owned by crm-backend; we +-- mirror only the -- minimum columns the billing suite needs, using identical column names so the -- CrmCoreReadPort queries run unchanged against the production CRM DB. @@ -149,7 +150,12 @@ CREATE INDEX IF NOT EXISTS idx_meter_consumption_lookup CREATE TABLE IF NOT EXISTS app_user ( id INTEGER GENERATED ALWAYS AS IDENTITY PRIMARY KEY, auth_user_id VARCHAR(255) NOT NULL UNIQUE, - email VARCHAR(256) NOT NULL + email VARCHAR(256) NOT NULL, + -- Preferred language, resolved onto every queued email at enqueue time. + locale VARCHAR(8) NULL, + -- Denormalised onto the queued row as the recipient display name. + first_name TEXT NULL, + last_name TEXT NULL ); -- ---- user_member_link (auth user ↔ member; for caller-scoped "my invoices") -- @@ -177,3 +183,75 @@ CREATE TABLE IF NOT EXISTS audit_log ( user_email VARCHAR(256), payload JSONB NOT NULL DEFAULT '{}'::jsonb ); + +-- ---- community_user (read by this service) --------------------------------- +-- Mirrors crm-backend's community_user join table. Read to narrow a +-- notification fan-out to a community's managers. +CREATE TABLE IF NOT EXISTS community_user ( + id_community INTEGER REFERENCES community (id) ON DELETE CASCADE, + id_user INTEGER REFERENCES app_user (id) ON DELETE CASCADE, + role VARCHAR(50) NOT NULL, + PRIMARY KEY (id_community, id_user) +); + +-- ---- notification (written by this service) -------------------------------- +-- Mirrors crm-backend's production DDL. This service only INSERTs one row per +-- recipient through core/notifications; reads are served by crm-backend. +CREATE TABLE IF NOT EXISTS notification ( + id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + id_community INTEGER REFERENCES community (id) ON DELETE CASCADE, + id_user INTEGER NOT NULL REFERENCES app_user (id) ON DELETE CASCADE, + type VARCHAR(128) NOT NULL, + data JSONB NOT NULL DEFAULT '{}'::jsonb, + read_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +-- ---- notification delivery layer (written / read by this service) ---------- +-- Mirrors crm-backend/database_script/2026-08-03_notification_delivery.sql. +-- `core/notifications` writes one outbound_message per emailable recipient and +-- reads notification_preference to decide what is deliverable. Sending and the +-- suppression check belong to the notification-dispatch worker; email_suppression +-- is mirrored here only so the schema stays a faithful copy. +CREATE TABLE IF NOT EXISTS outbound_message ( + id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY, + id_notification BIGINT NULL REFERENCES notification (id) ON DELETE SET NULL, + id_community INTEGER NULL REFERENCES community (id) ON DELETE CASCADE, + channel SMALLINT NOT NULL CHECK (channel IN (1, 2)), + recipient VARCHAR(320) NOT NULL, + recipient_name VARCHAR(255) NULL, + locale VARCHAR(8) NOT NULL DEFAULT '', + type VARCHAR(128) NOT NULL, + category SMALLINT NOT NULL CHECK (category IN (1, 2)), + data JSONB NOT NULL DEFAULT '{}'::jsonb, + dedupe_key VARCHAR(200) NOT NULL, + status SMALLINT NOT NULL DEFAULT 1 CHECK (status IN (1, 2, 3, 4, 5)), + attempts SMALLINT NOT NULL DEFAULT 0, + last_error TEXT NULL, + scheduled_for TIMESTAMPTZ NOT NULL DEFAULT NOW(), + claimed_at TIMESTAMPTZ NULL, + sent_at TIMESTAMPTZ NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); +CREATE UNIQUE INDEX IF NOT EXISTS uq_outbound_message_dedupe + ON outbound_message (dedupe_key); +CREATE INDEX IF NOT EXISTS ix_outbound_message_due + ON outbound_message (scheduled_for) WHERE status = 1; +CREATE INDEX IF NOT EXISTS ix_outbound_message_stale + ON outbound_message (claimed_at) WHERE status = 5; + +CREATE TABLE IF NOT EXISTS email_suppression ( + email VARCHAR(320) PRIMARY KEY, + reason SMALLINT NOT NULL CHECK (reason IN (1, 2, 3, 4)), + detail TEXT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE TABLE IF NOT EXISTS notification_preference ( + id_user INTEGER NOT NULL REFERENCES app_user (id) ON DELETE CASCADE, + type_prefix VARCHAR(128) NOT NULL, + channel SMALLINT NOT NULL CHECK (channel IN (1, 2)), + mode SMALLINT NOT NULL CHECK (mode IN (1, 3)), + + PRIMARY KEY (id_user, type_prefix, channel) +); diff --git a/tests/worker/test_worker_import_graph.py b/tests/worker/test_worker_import_graph.py new file mode 100644 index 0000000..1acc184 --- /dev/null +++ b/tests/worker/test_worker_import_graph.py @@ -0,0 +1,133 @@ +"""The worker image has no HTTP stack — pin that the worker never needs one. + +`Dockerfile.worker` installs `requirements/worker.txt`: base.txt plus the +numerical stack, and deliberately no fastapi, no uvicorn, no starlette. That +separation is the only reason a second Dockerfile exists. + +Its cost is that the constraint is invisible while developing, where `.venv` has +everything installed. An import added under `worker/` that reaches the API layer +type-checks, lints, and passes every test — then crash-loops the container. + +Which is exactly what happened: the overdue scheduler shipped importing +`api.billing.deps` for two provider functions, and `billing-worker` restarted 60+ +times in dev without the daily sweep ever running once. Nothing caught it, because +nothing here had ever asserted what the image contains. + +Two failure modes, one assertion each: + +1. the worker reaches a module importing something the image does not install; +2. the worker reaches a first-party package `Dockerfile.worker` does not COPY — + administrative-document's half of the same bug (`No module named 'api'`), which + no import check can see, because the package is present locally. + +Both need a subprocess: this pytest process has fastapi loaded already, and the +graph is only visible on a from-scratch import. +""" + +from __future__ import annotations + +import json +import os +import subprocess # a fresh interpreter is the only way to get a clean sys.modules +import sys +from pathlib import Path + +import pytest + +SERVICE_ROOT = Path(__file__).resolve().parents[2] + +# worker.main imports the other two, but naming them keeps a failure pointing at +# the guilty module rather than at the entry point. +WORKER_MODULES = ["worker.main", "worker.scheduler", "worker.sweeps"] + +# Top-level distributions `requirements/worker.txt` does not install. The first +# three come with fastapi/uvicorn; `jose` and `auth0` are api.txt-only too, and +# would be just as fatal. +ABSENT_FROM_THE_WORKER_IMAGE = ["fastapi", "starlette", "uvicorn", "jose", "auth0"] + +_PROBE_BODY = ''' +import importlib +import json +import pathlib +import sys + + +class NotInTheWorkerImage: + """Refuses BLOCKED packages from sys.meta_path, as the image does.""" + + def find_spec(self, name, path=None, target=None): + if name.partition(".")[0] in BLOCKED: + raise ImportError(f"No module named {name!r} — not in the worker image") + return None + + +sys.meta_path.insert(0, NotInTheWorkerImage()) + +for module in MODULES: + importlib.import_module(module) + +# Then report which first-party packages the graph actually touched, so the +# caller can check them against the Dockerfile's COPY lines. +root = pathlib.Path.cwd().resolve() +packages = set() +for module in list(sys.modules.values()): + origin = getattr(module, "__file__", None) + if not origin: + continue + try: + top = pathlib.Path(origin).resolve().relative_to(root).parts[0] + except ValueError: + continue # a dependency from site-packages, not ours + if (root / top / "__init__.py").exists(): + packages.add(top) + +print(json.dumps(sorted(packages))) +''' + +_PROBE = f"BLOCKED = {ABSENT_FROM_THE_WORKER_IMAGE!r}\nMODULES = {WORKER_MODULES!r}\n{_PROBE_BODY}" + + +@pytest.fixture(scope="module") +def probe() -> subprocess.CompletedProcess[str]: + """Import the worker in a fresh interpreter with the API stack blocked.""" + environment = os.environ.copy() + environment.setdefault("ENV", "test") + return subprocess.run( # noqa: S603 — fixed argv, no shell, sys.executable + [sys.executable, "-c", _PROBE], + cwd=SERVICE_ROOT, + capture_output=True, + text=True, + env=environment, + check=False, + ) + + +def _packages_copied_into_the_image() -> set[str]: + """The COPY sources in Dockerfile.worker, as bare directory names.""" + copied = set() + dockerfile = (SERVICE_ROOT / "Dockerfile.worker").read_text(encoding="utf-8") + for line in dockerfile.splitlines(): + fields = line.strip().split() + if not fields or fields[0] != "COPY" or any(f.startswith("--") for f in fields): + continue + copied.add(fields[1].rstrip("/")) + return copied + + +def test_the_worker_imports_with_the_api_stack_uninstalled(probe): + assert probe.returncode == 0, ( + "A worker module reaches something the worker image does not install. " + "Move the shared piece somewhere framework-free (see ports/providers.py) " + f"rather than adding it to requirements/worker.txt.\n\n{probe.stderr}" + ) + + +def test_every_package_the_worker_imports_is_copied_into_the_image(probe): + assert probe.returncode == 0, probe.stderr + imported = set(json.loads(probe.stdout.strip().splitlines()[-1])) + assert imported, "the probe reported no first-party packages, so it proved nothing" + missing = imported - _packages_copied_into_the_image() + assert not missing, ( + f"Dockerfile.worker has no COPY for {sorted(missing)}, which the worker " + "imports. The container would die at startup with ModuleNotFoundError." + ) diff --git a/worker/docgen_results.py b/worker/docgen_results.py index 26b0513..a8ebedc 100644 --- a/worker/docgen_results.py +++ b/worker/docgen_results.py @@ -30,6 +30,24 @@ _NAK_RETRY_DELAY_SECONDS = 30 +def _tenant_id_from(body: dict) -> str | int | None: + """Read the tenant from a docgen result. + + ``GenerationResult`` is ``extra="forbid"`` and has NO ``tenant_id`` field — + it exists only on the *request*. The one channel that round-trips is + ``metadata``, which document-generation echoes verbatim, so that is where the + tenant actually arrives. The top-level read is kept first only so a hand-built + payload (older tests, a manual replay) still works. + """ + top_level = body.get("tenant_id") + if top_level is not None: + return top_level # type: ignore[no-any-return] # untyped JSON body + metadata = body.get("metadata") + if not isinstance(metadata, dict): + return None + return metadata.get("tenant_id") + + async def process_docgen_result( body: dict, *, @@ -45,7 +63,7 @@ async def process_docgen_result( crm = crm_session or AsyncSessionCRMFactory() try: request_id = body.get("request_id") - tenant_id = body.get("tenant_id") + tenant_id = _tenant_id_from(body) if not request_id or tenant_id is None: logger.error("docgen result missing request_id/tenant_id: %r", body) return "drop" diff --git a/worker/issue.py b/worker/issue.py index 443d5dd..061525c 100644 --- a/worker/issue.py +++ b/worker/issue.py @@ -107,7 +107,13 @@ async def process_issue( reply_to=settings.DOCGEN_RESULT_SUBJECT, locale=settings.DEFAULT_LOCALE, presign_ttl=settings.DOCGEN_PRESIGN_TTL, - metadata={"invoice_id": invoice_id}, + # tenant_id rides in metadata because that is the ONLY field + # document-generation echoes back: GenerationResult is + # extra="forbid" and declares no tenant_id of its own. + metadata={ + "invoice_id": invoice_id, + "tenant_id": str(invoice.id_community), + }, ) ) return True diff --git a/worker/main.py b/worker/main.py index cc45a17..016ff6a 100644 --- a/worker/main.py +++ b/worker/main.py @@ -26,6 +26,7 @@ from core.tracing import setup_tracer_provider from regime.registry import assert_regime_parity from worker import dispatcher +from worker.scheduler import run_overdue_scheduler logger = logging.getLogger(__name__) @@ -104,9 +105,17 @@ async def main() -> None: inflight: set[asyncio.Task] = set() subs: list = [] heartbeat_task: asyncio.Task | None = None + scheduler_task: asyncio.Task | None = None try: subs = await dispatcher.subscribe_all(js, inflight=inflight) heartbeat_task = asyncio.create_task(_heartbeat(shutdown_event), name="heartbeat") + # The overdue sweep. It lives here rather than behind a cron container + # because it needs the service, its two sessions and the tenant + # ContextVar — all of which this process already has. An advisory lock + # inside makes a second replica a no-op. + scheduler_task = asyncio.create_task( + run_overdue_scheduler(shutdown_event), name="overdue-scheduler" + ) logger.info("Billing worker ready — listening on the billing queues") await shutdown_event.wait() @@ -117,6 +126,11 @@ async def main() -> None: with contextlib.suppress(asyncio.CancelledError, Exception): await heartbeat_task + if scheduler_task is not None: + scheduler_task.cancel() + with contextlib.suppress(asyncio.CancelledError, Exception): + await scheduler_task + # Let in-flight handlers finish + ack while NATS is still up. Stragglers # redeliver after ack_wait (the persistence idempotency guard makes that # safe), so bound the wait to stay under the SIGTERM grace period. diff --git a/worker/persistence.py b/worker/persistence.py index eeef2f8..17f00ec 100644 --- a/worker/persistence.py +++ b/worker/persistence.py @@ -16,6 +16,14 @@ from core import metrics as app_metrics from core.audit_log import AuditActions, AuditLogInput, AuditLogService from core.database.database import AsyncSessionCRMFactory, AsyncSessionLocalFactory +from core.notifications import ( + MANAGER_ROLES, + Channel, + CommunityTarget, + NotificationCategory, + NotificationService, + NotificationTypes, +) from regime.registry import RegimeConfigError, get_registry from shared.const import BillingDirection, BillingRunStatus, TariffKind from shared.models.local_models import BillingRunModel @@ -109,6 +117,18 @@ async def process_billing_run( ), id_community=run.id_community, ) + # Same CRM transaction as the audit row. `with_tenant` has already exited + # here, so the community is passed explicitly — exactly as the audit call + # above does, and core/notifications never reads a ContextVar. Redelivery + # safety is free: both early-return paths return from inside the + # `with_tenant` block, skipping the audit and this notification alike. + await NotificationService(crm).publish( + type=NotificationTypes.BILLING_RUN_COMPLETED, + target=CommunityTarget(community_id=run.id_community, roles=MANAGER_ROLES), + category=NotificationCategory.INFORMATIONAL, + channels=(Channel.INAPP,), + data={"run_id": run_id, "invoice_count": count}, + ) if own_crm: await crm.commit() return count diff --git a/worker/scheduler.py b/worker/scheduler.py new file mode 100644 index 0000000..b0e237c --- /dev/null +++ b/worker/scheduler.py @@ -0,0 +1,59 @@ +"""A once-a-day tick for the overdue sweep. + +Fires at a fixed LOCAL time rather than "every N hours from process start". A +from-start interval re-runs on every deploy, and the sweep is user-visible: +`invoice.overdue` is TRANSACTIONAL and reaches the member's inbox. Re-running is +harmless (an already-OVERDUE invoice matches nothing), so this is hygiene rather +than correctness — but it is the difference between one notice and one per +deploy. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import datetime +import logging +from zoneinfo import ZoneInfo + +from core.config import settings +from worker.sweeps import try_sweep_overdue + +logger = logging.getLogger(__name__) + +_SETTLEMENT_TZ = ZoneInfo("Europe/Brussels") + + +def seconds_until_next_run(now: datetime.datetime, hour_local: int) -> float: + """Seconds from ``now`` to the next occurrence of ``hour_local`` in Brussels. + + Pure, so the wrap-around and DST cases are unit-testable without waiting a + day. ``now`` must be timezone-aware. + """ + local_now = now.astimezone(_SETTLEMENT_TZ) + target = local_now.replace(hour=hour_local, minute=0, second=0, microsecond=0) + if target <= local_now: + target += datetime.timedelta(days=1) + return (target - local_now).total_seconds() + + +async def run_overdue_scheduler(shutdown: asyncio.Event) -> None: + """Sleep until the next scheduled hour, sweep, repeat, until shutdown.""" + if not settings.OVERDUE_SWEEP_ENABLED: + logger.info("overdue scheduler disabled") + return + while not shutdown.is_set(): + delay = seconds_until_next_run( + datetime.datetime.now(datetime.UTC), settings.OVERDUE_SWEEP_HOUR_LOCAL + ) + logger.info("next overdue sweep in %.0fs", delay) + with contextlib.suppress(TimeoutError): + await asyncio.wait_for(shutdown.wait(), timeout=delay) + if shutdown.is_set(): + return + try: + await try_sweep_overdue() + except Exception: + # One failed sweep must not take the worker down: the next tick + # retries, and the sweep is idempotent by construction. + logger.exception("overdue sweep failed") diff --git a/worker/sweeps.py b/worker/sweeps.py new file mode 100644 index 0000000..3af1bbb --- /dev/null +++ b/worker/sweeps.py @@ -0,0 +1,147 @@ +"""The scheduled overdue sweep. + +`POST /billing-runs/overdue-sweep` has existed since invoicing landed and +nothing ever invoked it, so no invoice was ever marked overdue on its own and +`invoice.overdue` — which is TRANSACTIONAL and goes to the member's inbox — +could not fire. + +This module is that scheduler. It calls the same `BillingService` the route +calls, so there is exactly one implementation of the sweep. +""" + +from __future__ import annotations + +import datetime +import logging +from zoneinfo import ZoneInfo + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from api.billing.repository import BillingRepository +from api.billing.service import BillingService +from core.audit_log import AuditLogService +from core.config import settings +from core.database.database import AsyncSessionCRMFactory, AsyncSessionLocalFactory +from ports.crm_core_sqlalchemy import SqlAlchemyCrmCoreRead + +# `ports.providers`, NOT `api.billing.deps`: the providers are the same objects, +# but `deps` imports fastapi, which `requirements/worker.txt` does not install — +# so importing it here crash-loops the worker image on startup. +from ports.providers import get_email_port, get_event_publisher +from regime.registry import get_registry +from shared.const import InvoiceStatus +from shared.models.local_models import InvoiceModel +from worker.context import with_tenant + +logger = logging.getLogger(__name__) + +_SETTLEMENT_TZ = ZoneInfo("Europe/Brussels") + +# One advisory lock for the whole sweep, so two worker replicas cannot both run +# it. `dedupe_key` would collapse the duplicate EMAIL, but nothing collapses the +# duplicate in-app notification. Namespaced by service; must stay stable. +_SWEEP_ADVISORY_LOCK_KEY = 0x0B111_0001 + + +async def _communities_with_overdue_invoices_unscoped( + local: AsyncSession, *, today: datetime.date +) -> list[int]: + """Which communities have an invoice the sweep would flip. + + **Deliberately unscoped**, hence the name. Every other owned-DB read goes + through `with_community_scope`, which filters on the + `current_internal_community_id` ContextVar — but a scheduler has no request + and therefore no tenant yet, and a scoped read here would match nothing with + no error, which is indistinguishable from "no work to do". + """ + stmt = ( + select(InvoiceModel.id_community) + .where( + InvoiceModel.status.in_([InvoiceStatus.ISSUED, InvoiceStatus.SENT]), + InvoiceModel.due_date < today, + ) + .distinct() + ) + return list((await local.execute(stmt)).scalars().all()) + + +async def sweep_overdue_for_every_community( + *, + local_session: AsyncSession | None = None, + crm_session: AsyncSession | None = None, + today: datetime.date | None = None, +) -> tuple[int, int]: + """Run the overdue sweep for every community with something to flip. + + Sessions are injectable, mirroring `process_billing_run`, so a test can drive + this on a rolled-back session without a container. + Returns (invoices marked, communities swept). + """ + own_local = local_session is None + own_crm = crm_session is None + local = local_session or AsyncSessionLocalFactory() + crm = crm_session or AsyncSessionCRMFactory() + as_of = today or datetime.datetime.now(_SETTLEMENT_TZ).date() + try: + community_ids = await _communities_with_overdue_invoices_unscoped(local, today=as_of) + if not community_ids: + return 0, 0 + + # The real service, not a reimplementation: `sweep_overdue` uses none of + # `registry`/`publisher`/`email`, but constructing the genuine article is + # what keeps the route and the scheduler on one code path. + service = BillingService( + local_session=local, + crm_session=crm, + repository=BillingRepository(local), + crm_read=SqlAlchemyCrmCoreRead(crm), + registry=get_registry(), + publisher=get_event_publisher(), + email=get_email_port(), + audit=AuditLogService(crm), + settings=settings, + ) + marked_total = 0 + swept = 0 + for id_community in community_ids: + # Without this the sweep's UPDATE filters on a None community and + # silently matches nothing. + with with_tenant(id_community): + result = await service.sweep_overdue() + marked_total += result.marked + swept += 1 + logger.info( + "overdue sweep: %s communit(ies), %s invoice(s) marked", + swept, + marked_total, + extra={"operation": "worker:overdue_sweep"}, + ) + return marked_total, swept + finally: + if own_local: + await local.close() + if own_crm: + await crm.close() + + +async def try_sweep_overdue() -> bool: + """Take the advisory lock and sweep. Returns False if another replica has it. + + `pg_try_advisory_lock` is session-scoped, so the lock lives exactly as long + as this connection and is released even if the process dies — the property a + cron-style lock needs and a lock table does not have. + """ + from sqlalchemy import func + + async with AsyncSessionLocalFactory() as local: + acquired = await local.scalar(select(func.pg_try_advisory_lock(_SWEEP_ADVISORY_LOCK_KEY))) + if not acquired: + logger.info("overdue sweep skipped: another replica holds the lock") + return False + try: + async with AsyncSessionCRMFactory() as crm: + await sweep_overdue_for_every_community(local_session=local, crm_session=crm) + return True + finally: + await local.execute(select(func.pg_advisory_unlock(_SWEEP_ADVISORY_LOCK_KEY)))