Notifications¶

Carbon Connect provides both email notifications (via AWS SES) and in-app notifications (database-backed).

Email Service¶

Source: backend/app/services/email_service.py

Overview¶

The email service sends transactional emails using AWS Simple Email Service (SES). It supports template-based rendering with HTML and plain-text variants.

Features¶

• AWS SES integration for reliable email delivery
• Template system with HTML and plain-text variants
• Batch sending for notifications to multiple recipients
• Rate limiting via Celery queue (30 emails/minute)
• Retry logic with exponential backoff (up to 30 minutes, 5 retries)
• SMTP fallback for development environments

Email Templates¶

The service includes templates for common notification types:

Usage¶

from backend.app.services.email_service import EmailService, get_email_service

email_service = get_email_service()

# Send a single email
await email_service.send_email(
    to="user@example.com",
    subject="New Grant Match Found",
    template="new_match",
    context={
        "user_name": "John",
        "grant_title": "Horizon Europe Clean Energy",
        "match_score": 0.85,
    },
)

# Send batch notifications
await email_service.send_batch(
    recipients=["user1@example.com", "user2@example.com"],
    subject="Deadline Approaching",
    template="deadline_approaching",
    context={"grant_title": "ERDF Energy Efficiency"},
)

Configuration¶

Notification Service¶

Source: backend/app/services/notification_service.py

Overview¶

The in-app notification service manages user notifications stored in the database. Notifications appear in the dashboard and can be marked as read or dismissed.

Notification Model¶

Notification Types¶

Usage¶

from backend.app.services.notification_service import NotificationService

service = NotificationService(db_session)

# Create a notification
await service.create_notification(
    user_id=user.id,
    tenant_id=user.tenant_id,
    type="new_match",
    title="New Grant Match",
    message="A new grant matching your profile was found.",
    data={"grant_id": str(grant.id), "score": 0.85},
)

# Get unread notifications
notifications = await service.get_unread(user_id=user.id)

# Mark as read
await service.mark_read(notification_id=notification.id)

# Mark all as read
await service.mark_all_read(user_id=user.id)

API Endpoints¶

Celery Integration¶

Email and notification tasks are routed to the email queue:

Rate limit: 30 emails per minute to avoid spam filter issues.

flowchart TB
    A[Celery Beat<br/>Hourly Check] --> B[Check Deadlines]
    B --> C{Approaching<br/>Deadlines?}
    C -->|Yes| D[Create Notifications]
    D --> E[Send Emails<br/>via SES]
    C -->|No| F[Skip]