Notification System (SMS, Email, In-App Realtime)

[borhanst]

Description

Implement a notification system for the admin kit that supports 3 notification channels:

• SMS (default, supported out of the box)
• Email (default, supported out of the box)
• In-App Realtime (needs configuration/setup, WebSocket/SSE based)

Requirements

Core Notification Service

• Unified notification service that abstracts the channel (SMS, Email, In-App)
• Support for multiple channels simultaneously (e.g., send both Email + SMS)
• Configurable per-notification which channels to use
• Fallback mechanism if one channel fails

Channels

1. Email — Built-in support via SMTP or transactional email provider
2. SMS — Built-in support via Twilio
   • Base class for SMS providers with abstract interface
   • Twilio as the first built-in implementation
   • Users can create custom SMS providers by extending the base class
3. In-App Realtime — Notifications pushed to client in real-time via WebSocket/SSE
   • Requires additional setup (WebSocket server, notification preferences, UI components)
   • Notifications stored in DB for history, but delivered in real-time to connected clients

SMS Provider Architecture

sms/
├── base.py          # Abstract base class (SMSProvider)
├── twilio.py        # Twilio implementation
├── custom/          # User-created providers
│   └── example.py
└── __init__.py

Base SMS Provider

class SMSProvider(ABC):
    @abstractmethod
    async def send(self, to: str, message: str) -> SMSResult:
        """Send SMS to a phone number"""
        pass

    @abstractmethod
    async def check_status(self, message_id: str) -> SMSStatus:
        """Check delivery status"""
        pass

Twilio Implementation

class TwilioSMSProvider(SMSProvider):
    def __init__(self, account_sid: str, auth_token: str, from_number: str):
        ...

    async def send(self, to: str, message: str) -> SMSResult:
        # Twilio API call
        pass

    async def check_status(self, message_id: str) -> SMSStatus:
        # Check via Twilio API
        pass

Custom Provider (User-Defined)

class MyCustomSMSProvider(SMSProvider):
    async def send(self, to: str, message: str) -> SMSResult:
        # Call any SMS API (Vonage, AWS SNS, custom gateway, etc.)
        pass

    async def check_status(self, message_id: str) -> SMSStatus:
        # Custom status check
        pass

# Register it
notification_service.register_sms_provider("custom", MyCustomSMSProvider(...))

Realtime In-App Notifications

• WebSocket or Server-Sent Events (SSE) connection per user
• Instant delivery when notification is created — no polling
• Fallback to polling if WebSocket connection drops
• Notification badge/count updates in real-time
• Connection management: reconnect handling, heartbeat/ping

Standalone / Reusable

• The notification system must be usable outside of the admin panel
• Users can integrate it into their custom routes and services
• Provide a clean API/interface for programmatic usage
• Example: notify(user_id, message, channels=["sms", "email"])

Notification Features

• Send single or batch notifications
• Notification templates (email templates, SMS templates)
• Notification history/logs
• User notification preferences (opt-in/opt-out per channel)
• Read/unread status for in-app notifications

API Endpoints

• POST /notifications/send — Send notification
• GET /notifications — List user notifications (in-app)
• PUT /notifications/{id}/read — Mark as read
• PUT /notifications/preferences — Update channel preferences
• WS /notifications/ws — WebSocket endpoint for real-time notifications
• GET /notifications/stream — SSE fallback endpoint

Acceptance Criteria

• Notification service is a standalone module that can be imported independently
• SMS provider base class with abstract interface
• Twilio SMS provider implementation
• Users can create custom SMS providers by extending base class
• SMS and Email work out of the box with minimal config
• In-App notifications are delivered in real-time via WebSocket
• SSE fallback when WebSocket is unavailable
• Users can call the service from any custom route
• Notification templates are configurable
• Batch sending is supported
• Notification preferences are stored per user
• Reconnection handling for dropped WebSocket connections

Additional Notes

• Design for extensibility — users may want to add Push Notifications, WhatsApp, etc. later
• Use a provider-based pattern so channels are pluggable
• Keep the default setup simple but allow advanced customization
• Consider using FastAPI WebSocket or SSE library for real-time delivery