docs/dev/NOTIFICATION_SYSTEM.md

Notification System - Developer Guide

🎯 Overview

The SwiftOps notification system is a 2-tier architecture designed for reliability, scalability, and maintainability. It handles all notifications across the platform with a consistent, world-class approach.

Architecture

┌─────────────────────────────────────────────────────────────┐
│                    NOTIFICATION SYSTEM                       │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│  TIER 1: Notification Creation (Synchronous)                │
│  ┌────────────────────────────────────────────────────┐    │
│  │  NotificationCreator                                │    │
│  │  - Creates notification records in database         │    │
│  │  - Synchronous, transaction-safe                    │    │
│  │  - Guaranteed to be saved                           │    │
│  │  - Rolls back with parent operation                 │    │
│  └────────────────────────────────────────────────────┘    │
│                          ↓                                   │
│  TIER 2: Notification Delivery (Asynchronous)               │
│  ┌────────────────────────────────────────────────────┐    │
│  │  NotificationDelivery                               │    │
│  │  - Delivers via external channels (WhatsApp, etc.)  │    │
│  │  - Non-blocking background tasks                    │    │
│  │  - Handles failures gracefully                      │    │
│  │  - Configurable per channel                         │    │
│  └────────────────────────────────────────────────────┘    │
│                                                              │
└─────────────────────────────────────────────────────────────┘

---

🚀 Quick Start

Basic Usage

from fastapi import BackgroundTasks
from app.services.notification_creator import NotificationCreator
from app.services.notification_delivery import NotificationDelivery

# In your endpoint
@router.post("/tickets/assign")
def assign_ticket(
    background_tasks: BackgroundTasks,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user)
):
    # Your business logic
    ticket = create_ticket(...)
    assignment = assign_to_agent(...)
    
    # TIER 1: Create notification (synchronous)
    notification = NotificationCreator.create(
        db=db,
        user_id=agent.id,
        title="New Ticket Assigned",
        message=f"You have been assigned to {ticket.ticket_name}",
        source_type="ticket",
        source_id=ticket.id,
        notification_type="assignment",
        channel="whatsapp",
        project_id=ticket.project_id,
        metadata={
            "ticket_number": ticket.ticket_number,
            "priority": "high",
            "action_url": f"/tickets/{ticket.id}"
        }
    )
    
    # Commit notification with your business logic
    db.commit()
    
    # TIER 2: Queue delivery (asynchronous, non-blocking)
    NotificationDelivery.queue_delivery(
        background_tasks=background_tasks,
        notification_id=notification.id
    )
    
    return {"status": "success"}

---

📚 API Reference

NotificationCreator

create() - Create Single Notification

notification = NotificationCreator.create(
    db: Session,                    # Database session
    user_id: UUID,                  # User to notify
    title: str,                     # Short title (for display)
    message: str,                   # Detailed message
    source_type: str,               # Entity type (ticket, expense, payroll, etc.)
    source_id: Optional[UUID],      # Entity ID (None for bulk operations)
    notification_type: str,         # Notification type (assignment, payment, alert, etc.)
    channel: str = "in_app",        # Delivery channel (in_app, whatsapp, email, sms, push)
    metadata: Optional[Dict] = None,# Additional data (action URLs, context, etc.)
    project_id: Optional[UUID] = None  # Optional project ID for filtering
) -> Notification

Example:

notification = NotificationCreator.create(
    db=db,
    user_id=agent.id,
    title="Ticket Assigned",
    message="You have been assigned to install fiber at Customer A",
    source_type="ticket",
    source_id=ticket.id,
    notification_type="assignment",
    channel="whatsapp",
    project_id=ticket.project_id,
    metadata={
        "ticket_number": "TKT-001",
        "priority": "high",
        "action_url": f"/tickets/{ticket.id}"
    }
)
db.commit()

create_bulk() - Create Multiple Notifications

notifications = NotificationCreator.create_bulk(
    db: Session,
    user_ids: List[UUID],           # List of users to notify
    title: str,                     # Same title for all
    message: str,                   # Same message for all
    source_type: str,
    source_id: Optional[UUID],
    notification_type: str,
    channel: str = "in_app",
    metadata: Optional[Dict] = None,
    project_id: Optional[UUID] = None
) -> List[Notification]

Example:

# Notify all workers about payroll export
worker_ids = [worker1.id, worker2.id, worker3.id]
notifications = NotificationCreator.create_bulk(
    db=db,
    user_ids=worker_ids,
    title="💰 Payment Processed",
    message="Your payment has been processed",
    source_type="payroll",
    source_id=None,
    notification_type="payment",
    channel="whatsapp"
)
db.commit()

notify_project_team() - Notify Team Members by Role

notifications = NotificationCreator.notify_project_team(
    db: Session,
    project_id: UUID,               # Project ID
    title: str,
    message: str,
    source_type: str,
    source_id: Optional[UUID],
    notification_type: str,
    roles: Optional[List[AppRole]] = None,  # Filter by roles (PM, Dispatcher, etc.)
    channel: str = "in_app",
    metadata: Optional[Dict] = None,
    exclude_user_ids: Optional[List[UUID]] = None  # Exclude specific users
) -> List[Notification]

Example:

# Notify all managers when ticket is dropped
notifications = NotificationCreator.notify_project_team(
    db=db,
    project_id=ticket.project_id,
    title="⚠️ Ticket Dropped - Action Required",
    message=f"{agent.name} dropped ticket: {ticket.name}",
    source_type="ticket",
    source_id=ticket.id,
    notification_type="ticket_dropped",
    roles=[AppRole.PROJECT_MANAGER, AppRole.DISPATCHER],
    exclude_user_ids=[agent.id]  # Don't notify the agent who dropped
)
db.commit()

---

NotificationDelivery

queue_delivery() - Queue Single Notification

NotificationDelivery.queue_delivery(
    background_tasks: BackgroundTasks,  # FastAPI BackgroundTasks
    notification_id: UUID               # Notification ID to deliver
) -> None

Example:

NotificationDelivery.queue_delivery(
    background_tasks=background_tasks,
    notification_id=notification.id
)

queue_bulk_delivery() - Queue Multiple Notifications

NotificationDelivery.queue_bulk_delivery(
    background_tasks: BackgroundTasks,
    notification_ids: List[UUID]        # List of notification IDs
) -> None

Example:

NotificationDelivery.queue_bulk_delivery(
    background_tasks=background_tasks,
    notification_ids=[n.id for n in notifications]
)

---

🎨 Common Patterns

Pattern 1: Single User Notification

@router.post("/tickets/assign")
def assign_ticket(
    background_tasks: BackgroundTasks,
    db: Session = Depends(get_db)
):
    # Business logic
    assignment = create_assignment(...)

    # Create notification
    notification = NotificationCreator.create(
        db=db,
        user_id=agent.id,
        title="Ticket Assigned",
        message=f"You have been assigned to {ticket.name}",
        source_type="ticket",
        source_id=ticket.id,
        notification_type="assignment",
        channel="whatsapp"
    )
    db.commit()

    # Queue delivery
    NotificationDelivery.queue_delivery(
        background_tasks=background_tasks,
        notification_id=notification.id
    )

    return assignment

Pattern 2: Notify Project Team by Role

@router.post("/tickets/{ticket_id}/drop")
def drop_ticket(
    background_tasks: BackgroundTasks,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user)
):
    # Business logic
    ticket = drop_ticket(...)

    # Notify all managers and dispatchers
    notifications = NotificationCreator.notify_project_team(
        db=db,
        project_id=ticket.project_id,
        title="⚠️ Ticket Dropped",
        message=f"{current_user.name} dropped ticket: {ticket.name}",
        source_type="ticket",
        source_id=ticket.id,
        notification_type="ticket_dropped",
        roles=[AppRole.PROJECT_MANAGER, AppRole.DISPATCHER],
        exclude_user_ids=[current_user.id]
    )
    db.commit()

    # Queue delivery
    NotificationDelivery.queue_bulk_delivery(
        background_tasks=background_tasks,
        notification_ids=[n.id for n in notifications]
    )

    return ticket

---

⚙️ Configuration

Channel Configuration

For MVP, all external channels are disabled by default. Notifications are created but not delivered externally.

File: src/app/services/notification_delivery.py

class NotificationDeliveryConfig:
    # For MVP, all external channels are disabled
    ENABLE_WHATSAPP = False  # Set to True when ready
    ENABLE_EMAIL = False     # Set to True when ready
    ENABLE_SMS = False       # Set to True when ready
    ENABLE_PUSH = False      # Set to True when ready

    # In-app notifications are always enabled
    ENABLE_IN_APP = True

To enable a channel:

1. Set the flag to True in NotificationDeliveryConfig
2. Implement the delivery method (e.g., _deliver_whatsapp())
3. Test thoroughly
4. Deploy

---

🔍 Best Practices

1. Always Commit Notifications

# ✅ CORRECT
notification = NotificationCreator.create(...)
db.commit()  # Commit before queuing delivery

NotificationDelivery.queue_delivery(...)

# ❌ WRONG
notification = NotificationCreator.create(...)
NotificationDelivery.queue_delivery(...)  # Notification not committed yet!
db.commit()

2. Use Metadata for Context

# ✅ CORRECT - Rich metadata
notification = NotificationCreator.create(
    db=db,
    user_id=user.id,
    title="Ticket Assigned",
    message="You have been assigned...",
    source_type="ticket",
    source_id=ticket.id,
    notification_type="assignment",
    metadata={
        "ticket_number": ticket.ticket_number,
        "priority": "high",
        "action_url": f"/tickets/{ticket.id}",
        "customer_name": ticket.customer_name
    }
)

3. Handle Notification Failures Gracefully

# ✅ CORRECT - Don't fail business logic if notification fails
try:
    notification = NotificationCreator.create(...)
    db.commit()
    NotificationDelivery.queue_delivery(...)
except Exception as e:
    logger.error(f"Failed to create notification: {e}")
    # Continue with business logic

---

🚀 Migration from Old System

Old Pattern (NotificationHelper - Async)

# ❌ OLD - Don't use this anymore
await NotificationHelper.notify_ticket_assigned(
    db=db,
    ticket=ticket,
    agent=agent
)

New Pattern (NotificationCreator - Sync)

# ✅ NEW - Use this instead
notification = NotificationCreator.create(
    db=db,
    user_id=agent.id,
    title="Ticket Assigned",
    message=f"You have been assigned to {ticket.name}",
    source_type="ticket",
    source_id=ticket.id,
    notification_type="assignment",
    channel="whatsapp"
)
db.commit()

NotificationDelivery.queue_delivery(
    background_tasks=background_tasks,
    notification_id=notification.id
)

---

📞 Support

For questions or issues with the notification system:

1. Check this documentation
2. Review existing implementations in:
   • src/app/api/v1/payroll.py (payroll export)
   • src/app/api/v1/ticket_assignments.py (ticket drop)
   • src/app/services/expense_service.py (expense submission)
3. Contact the development team

---

Last Updated: 2024-12-12 Version: 1.0.0 Status: Production Ready ✅