Push Notifications & Novu Integration Guide

This guide describes the push notification architecture for Framework M applications and documents production-proven patterns drawn from real-world implementations in tenant-registry and hornbill-m.

Architectural Philosophy: No Reinvention

Framework M intentionally does not implement in-tree APNs/FCM token management, device registration database tables (PushDevice), or client token invalidation feedback loops:

Separation of Concerns: Physical push notifications require device registry persistence, APNs/FCM certificate/credential rotation, multi-device delivery fanout, and user notification preference management (channel opt-ins/opt-outs, digests).
Delegation to Notification Infrastructure: Real-world push orchestration is best handled by dedicated notification engines such as Novu or specialized client SDKs.
Pluggable Hexagonal Ports: Framework M provides NotificationProtocol as a provider-neutral port. During local development and CI, standard adapters (ConsoleNotificationAdapter and NoopNotificationAdapter) satisfy the contract without external dependencies or runtime failures.

Production Proven Patterns: `tenant-registry` & `hornbill-m`

The developer patterns established across tenant-registry and hornbill-m demonstrate how to cleanly decouple applications from third-party notification services:

1. In `tenant-registry` (Composite Dispatch & Extensibility)

Port & Adapter: Uses a NotificationDispatcher port with a NovuNotificationAdapter (leveraging Novu's Event API).
Discovery via Entry Points: Registered via pyproject.toml entry points under framework_m.notification_dispatchers.
Dynamic DI Wiring: The TenantRegistryContainer discovers enabled dispatchers dynamically at resolution time and supports composite dispatching (e.g. Novu + SMTP email fallback).
Domain Decoupling: Domain services (such as InvitationService) invoke high-level dispatch methods like send_notification(recipient, template_id, payload) without any knowledge of Novu or transport specifics.

2. In `hornbill-m` (Zero-DB Push Tokens & httpx Gateway)

Zero DB Token Tables: Neither device push tokens nor delivery queues are stored in the application database. LocalUser.id acts directly as Novu's subscriberId.
Direct Client-Side Token Registration: Client apps register push tokens directly with Novu against the user's subscriberId:
- Web: Via @novu/notification-center.
- Mobile: Via Expo SDK / Novu mobile client SDKs.
Backend Trigger via httpx: Services invoke the notification gateway port:
```
await gateway.send_notification(subscriber_id, workflow_id, payload)
```
Implemented via a lightweight NovuAdapter using httpx directly to call Novu's /v1/events/trigger endpoint—avoiding SDK bloat and heavy third-party dependencies.
Orchestration Delegation: Novu handles device routing, multi-device fanout, in-app notification center feed, and channel fallbacks (e.g. push $\rightarrow$ email).
Dev & CI Safe Mode: If NOVU_API_KEY is unset or empty, the adapter safely logs as a mock and no-ops, ensuring tests and local development run reliably without network calls or external credentials.

Built-In Standard Adapters in Framework M

Framework M's standard package (framework-m-standard) includes two built-in notification adapters:

ConsoleNotificationAdapter (provider = "console"): Logs all notification events (push, email, SMS, in-app) to standard output via logging/structlog. Ideal for local development (m dev) and interactive debugging.
NoopNotificationAdapter (provider = "noop"): Silently accepts and discards payloads, returning True. Ideal for headless CI runs and deployments where notifications are disabled.

Configuration (`m.toml` or `framework_config.toml`)

[notifications]
provider = "console" # or "noop"

Or via environment variable:

FRAMEWORK_M_NOTIFICATIONS_PROVIDER=console

Reference Implementation: Production-Grade `NovuNotificationAdapter`

Below is the recommended reference adapter implementing NotificationProtocol using httpx with safe dev/CI mock mode:

from __future__ import annotations

import logging
from typing import Any
import httpx
from framework_m_core.interfaces.notification import NotificationProtocol

logger = logging.getLogger(__name__)


class NovuNotificationAdapter(NotificationProtocol):
    """Production Novu adapter matching the hornbill-m and tenant-registry architecture.
    
    Uses httpx directly without external SDK bloat. Safely falls back to no-op logging
    when NOVU_API_KEY is not configured.
    """

    def __init__(
        self,
        api_key: str | None = None,
        base_url: str = "https://api.novu.co/v1",
        timeout: float = 10.0,
    ) -> None:
        self.api_key = api_key
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout

    async def _trigger_workflow(
        self,
        workflow_id: str,
        subscriber_id: str,
        payload: dict[str, Any],
        overrides: dict[str, Any] | None = None,
    ) -> bool:
        """Trigger a Novu workflow via /v1/events/trigger."""
        # Safe mock mode for dev & CI when credentials are unset
        if not self.api_key:
            logger.info(
                "[NovuNotificationAdapter] Mock dispatch: workflow=%s subscriber=%s",
                workflow_id,
                subscriber_id,
                extra={"payload": payload},
            )
            return True

        body: dict[str, Any] = {
            "name": workflow_id,
            "to": {"subscriberId": subscriber_id},
            "payload": payload,
        }
        if overrides:
            body["overrides"] = overrides

        try:
            async with httpx.AsyncClient(timeout=self.timeout) as client:
                resp = await client.post(
                    f"{self.base_url}/events/trigger",
                    headers={
                        "Authorization": f"ApiKey {self.api_key}",
                        "Content-Type": "application/json",
                    },
                    json=body,
                )
                if resp.status_code in (200, 201):
                    return True
                logger.error(
                    "Novu event trigger failed (%s): %s",
                    resp.status_code,
                    resp.text,
                )
                return False
        except Exception as exc:
            logger.exception("Error contacting Novu API: %s", exc)
            return False

    async def send_push(
        self,
        user_id: str,
        title: str,
        body: str,
        *,
        data: dict[str, Any] | None = None,
    ) -> bool:
        """Dispatch push notification via Novu workflow."""
        workflow_id = (data or {}).get("workflow_id", "push-notification")
        payload = {
            "title": title,
            "body": body,
            **(data or {}),
        }
        return await self._trigger_workflow(workflow_id, user_id, payload)

    async def send_email(
        self,
        to: list[str],
        subject: str,
        body: str,
        *,
        template: str | None = None,
        context: dict[str, Any] | None = None,
        cc: list[str] | None = None,
        bcc: list[str] | None = None,
        attachments: list[tuple[str, bytes]] | None = None,
    ) -> bool:
        """Dispatch email notification via Novu workflow."""
        workflow_id = template or "email-notification"
        primary_recipient = to[0] if to else "unknown"
        payload = {
            "subject": subject,
            "body": body,
            "to": to,
            **(context or {}),
        }
        return await self._trigger_workflow(workflow_id, primary_recipient, payload)

    async def send_sms(
        self,
        to: str,
        body: str,
        *,
        template: str | None = None,
        context: dict[str, Any] | None = None,
    ) -> bool:
        """Dispatch SMS notification via Novu workflow."""
        workflow_id = template or "sms-notification"
        payload = {"body": body, "phone": to, **(context or {})}
        return await self._trigger_workflow(workflow_id, to, payload)

    async def send_in_app(
        self,
        user_id: str,
        title: str,
        body: str,
        *,
        link: str | None = None,
        data: dict[str, Any] | None = None,
    ) -> bool:
        """Dispatch in-app notification to Novu notification center."""
        workflow_id = (data or {}).get("workflow_id", "in-app-notification")
        payload = {
            "title": title,
            "body": body,
            "link": link,
            **(data or {}),
        }
        return await self._trigger_workflow(workflow_id, user_id, payload)

Registering the Custom Adapter in Framework M

To use the custom adapter, register it in your application's pyproject.toml:

[project.entry-points."framework_m.adapters.notification"]
default = "my_app.adapters.novu:NovuNotificationAdapter"

When Framework M boots, InitAdapters automatically discovers and binds your adapter to container.notification_adapter.

Architectural Philosophy: No Reinvention​

Production Proven Patterns: tenant-registry & hornbill-m​

1. In tenant-registry (Composite Dispatch & Extensibility)​

2. In hornbill-m (Zero-DB Push Tokens & httpx Gateway)​

Built-In Standard Adapters in Framework M​

Configuration (m.toml or framework_config.toml)​

Reference Implementation: Production-Grade NovuNotificationAdapter​

Registering the Custom Adapter in Framework M​