Notification features

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.
 

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/

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(

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:

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()

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"
 

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",
+]

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