Notification Flow - Complete Trace

Notification Flow - Complete Trace

This document traces the complete flow of notifications from research completion/failure through to delivery via Apprise.

Overview

Notifications are sent when:

1. Research completes successfully → RESEARCH_COMPLETED event
2. Research fails → RESEARCH_FAILED event
3. Research is queued → RESEARCH_QUEUED event
4. Subscription updates → SUBSCRIPTION_UPDATE event
5. Subscription errors → SUBSCRIPTION_ERROR event
6. API quota/rate limits exceeded → API_QUOTA_WARNING event
7. Authentication fails → AUTH_ISSUE event

Complete Flow: Research Completed

1. Research Thread Completes (research_service.py:1166-1168)

When research finishes successfully in the background thread:

# src/local_deep_research/web/services/research_service.py:1166
cleanup_research_resources(
    research_id, active_research, termination_flags, username
)

2. Cleanup Calls Queue Processor (research_service.py:1727)

The cleanup function notifies the queue processor:

# src/local_deep_research/web/services/research_service.py:1727
queue_processor.notify_research_completed(username, research_id, user_password=user_password)

Key Points:

• Called from background research thread
• Passes user_password for secure database access
• Uses processor_v2 which handles encrypted per-user databases

3. Queue Processor Updates Status (processor_v2.py:278-306)

# src/local_deep_research/web/queue/processor_v2.py:278-306
def notify_research_completed(self, username: str, research_id: str, user_password: str = None):
    with get_user_db_session(username, user_password) as session:
        # Update queue status
        queue_service = UserQueueService(session)
        queue_service.update_task_status(research_id, "completed")

        # Send notification if enabled
        self._send_research_notification(
            session=session,
            username=username,
            research_id=research_id,
            event_type="RESEARCH_COMPLETED",
        )

Key Points:

• Opens encrypted user database with password
• Updates queue status first
• Delegates to _send_research_notification helper

4. Build Notification Context (processor_v2.py:357-243)

The helper method prepares the notification:

# src/local_deep_research/web/queue/processor_v2.py:357-243
def _send_research_notification(
    self,
    session,
    username: str,
    research_id: str,
    event_type: str,
    error_message: str = None,
):
    # Get settings snapshot for thread-safe notification sending
    settings_manager = SettingsManager(session)
    settings_snapshot = settings_manager.get_settings_snapshot()

    # Lookup research details (with retry logic for timing issues)
    research = session.query(ResearchHistory).filter_by(id=research_id).first()

    if research:
        # Create notification manager with settings snapshot and user_id
        notification_manager = NotificationManager(
            settings_snapshot=settings_snapshot,
            user_id=username  # Enables per-user rate limiting
        )

        # Build full URL for notification
        full_url = build_notification_url(
            f"/research/{research_id}",
            settings_manager=settings_manager,
        )

        # Build notification context
        context = {
            "query": research.query or "Unknown query",
            "research_id": research_id,
            "summary": report_content[:200] + "...",  # Truncated
            "url": full_url,  # Full clickable URL
        }

        # Send notification (user_id already set in manager init)
        result = notification_manager.send_notification(
            event_type=EventType.RESEARCH_COMPLETED,
            context=context,
        )

Key Points:

• Settings Snapshot: Captures settings at notification time (thread-safe)
• Research Lookup: Queries database for research details with 3 retry attempts
• URL Building: Constructs full URL using app.external_url or app.host/app.port
• Context Building: Includes query, research_id, summary (truncated to 200 chars), full URL
• No Session Passed: NotificationManager gets settings_snapshot, NOT session (thread-safe)

5. NotificationManager Checks Settings (manager.py:91-126)

# src/local_deep_research/notifications/manager.py:91-126
def send_notification(
    self,
    event_type: EventType,
    context: Dict[str, Any],
    user_id: Optional[str] = None,
    force: bool = False,
) -> bool:
    # Check if notifications are enabled for this event type
    should_notify = self._should_notify(event_type, user_id)
    if not force and not should_notify:
        logger.debug(f"Notifications disabled for event type: {event_type.value}")
        return False

    # Check rate limit
    rate_limit_ok = self._rate_limiter.allow(user_id or "default")
    if not force and not rate_limit_ok:
        raise RateLimitError("Notification rate limit exceeded")

    # Get service URLs from settings snapshot
    service_urls = self._get_setting("notifications.service_url", default="")

    if not service_urls or not service_urls.strip():
        logger.debug("No notification service URLs configured")
        return False

    # Send notification with service URLs
    result = self.service.send_event(event_type, context, service_urls=service_urls)

    return result

Settings Checked (from settings_snapshot):

1. notifications.on_research_completed - Is this event type enabled? (default: False for most events)
2. Per-User Rate Limits - Check shared rate limiter with user-specific limits:
   • notifications.rate_limit_per_hour (default: 10) - Configured per user
   • notifications.rate_limit_per_day (default: 50) - Configured per user
   • Each user has independent rate limit counters
3. notifications.service_url - Comma-separated list of Apprise URLs (required)

Key Points:

• All settings come from settings_snapshot (captured earlier)
• Rate limiter is shared singleton with per-user limits and counters
• Each user's rate limits are configured independently when NotificationManager is created with user_id
• One user hitting their rate limit does NOT affect other users
• force=False by default (respects settings and rate limits)

6. NotificationService Formats Message (service.py:198-235)

# src/local_deep_research/notifications/service.py:198-235
def send_event(
    self,
    event_type: EventType,
    context: Dict[str, Any],
    service_urls: Optional[str] = None,
    tag: Optional[str] = None,
    custom_template: Optional[Dict[str, str]] = None,
) -> bool:
    # Format notification using template
    message = NotificationTemplate.format(
        event_type, context, custom_template
    )

    # Send notification
    result = self.send(
        title=message["title"],
        body=message["body"],
        service_urls=service_urls,
        tag=tag,
    )
    return result

Template Used (templates.py):

EventType.RESEARCH_COMPLETED: {
    "title": "Research Completed: {query}",
    "body": "Your research '{query}' has completed successfully.\n\n"
            "Summary: {summary}\n\n"
            "View results: {url}",
}

Context Variables:

• {query} - Research query text
• {summary} - Truncated report content (max 200 chars)
• {url} - Full clickable URL to view research
• {research_id} - Research ID (available but not used in default template)

7. NotificationService Sends via Apprise (service.py:47-196)

# src/local_deep_research/notifications/service.py:47-196
def send(
    self,
    title: str,
    body: str,
    service_urls: Optional[str] = None,
    tag: Optional[str] = None,
    attach: Optional[List[str]] = None,
) -> bool:
    # Retry logic with exponential backoff
    retry_delay = INITIAL_RETRY_DELAY  # 0.5s

    for attempt in range(1, MAX_RETRY_ATTEMPTS + 1):  # 3 attempts
        try:
            # Create temporary Apprise instance
            # Automatically garbage collected by Python when out of scope
            apprise_instance = apprise.Apprise()
            apprise_instance.add(service_urls, tag=tag)

            # Send notification
            notify_result = apprise_instance.notify(
                title=title,
                body=body,
                tag=tag,
                attach=attach,
            )

            if notify_result:
                return True

            # Retry with exponential backoff
            time.sleep(retry_delay)
            retry_delay *= RETRY_BACKOFF_MULTIPLIER  # 2x

        except Exception as e:
            logger.error(f"Error sending notification: {e}")
            time.sleep(retry_delay)
            retry_delay *= RETRY_BACKOFF_MULTIPLIER

    # All attempts failed
    raise SendError("Failed to send notification after 3 attempts")

Retry Strategy:

• Attempt 1: Immediate send
• Attempt 2: Wait 0.5s, retry
• Attempt 3: Wait 1.0s, retry
• After 3 attempts: Raise SendError

Apprise Instances:

• Temporary instances created for each send operation
• Automatically garbage collected by Python
• Multiple service URLs supported (comma-separated)

8. Apprise Delivers Notification

Apprise handles the actual delivery to configured services:

# User's settings: notifications.service_url
"discord://webhook_id/webhook_token,mailto://user:password@smtp.gmail.com"

Supported Services (via Apprise):

• Discord, Slack, Telegram
• Email (SMTP, Gmail, etc.)
• Pushover, Gotify
• Many more: https://github.com/caronc/apprise/wiki

Security:

• Service URLs encrypted at rest (AES-256 via SQLCipher)
• Encryption key derived from user's login password (PBKDF2-SHA512)
• URLs masked in logs (e.g., discord://webhook_id/***)

Complete Flow: Research Failed

Similar to completed flow, but:

1. Entry Point: research_service.py:1642 (exception handler)
2. Queue Method: queue_processor.queue_error_update() (processor.py)
3. Notification: Sent from processor.py:577-606 (error update handler)
4. Event Type: EventType.RESEARCH_FAILED
5. Context: Includes error field instead of summary

Template for RESEARCH_FAILED:

EventType.RESEARCH_FAILED: {
    "title": "Research Failed: {query}",
    "body": "Research on '{query}' failed.\n\n"
            "Error: {error}\n\n"
            "Please check the logs for more details.",
}

Note: Error messages are sanitized for security to avoid exposing sensitive information in notifications.

Settings Snapshot Pattern

Why Settings Snapshot?

Notifications are sent from background threads that shouldn't access Flask g or SQLite sessions (not thread-safe). The solution is to capture settings once in the main thread and pass them as a dict.

How It Works:

# 1. In main thread (with database session)
settings_manager = SettingsManager(session)
settings_snapshot = settings_manager.get_settings_snapshot()
# Returns dict like: {"notifications.service_url": "...", "notifications.on_research_completed": True, ...}

# 2. Pass to NotificationManager (thread-safe - no session!)
notification_manager = NotificationManager(
    settings_snapshot=settings_snapshot,
    user_id=username
)

# 3. NotificationManager reads from snapshot
def _get_setting(self, key: str, default: Any = None) -> Any:
    return self._settings_snapshot.get(key, default)

Benefits:

• ✅ Thread-safe (no database access in background threads)
• ✅ Consistent settings (captured at notification time, not changed mid-notification)
• ✅ No Flask g context needed
• ✅ Works from queue processors, schedulers, etc.

Rate Limiting

Implementation: In-memory, per-user limits and counters, shared singleton

class RateLimiter:
    _lock = threading.Lock()
    _user_limits: Dict[str, tuple[int, int]] = {}  # user_id -> (max_per_hour, max_per_day)
    _hourly_counts: Dict[str, deque] = {}  # user_id -> timestamps
    _daily_counts: Dict[str, deque] = {}   # user_id -> timestamps

    def set_user_limits(self, user_id: str, max_per_hour: int, max_per_day: int):
        """Configure rate limits for a specific user."""
        with self._lock:
            self._user_limits[user_id] = (max_per_hour, max_per_day)

    def allow(self, user_id: str) -> bool:
        with self._lock:  # Thread-safe
            now = datetime.now(timezone.utc)

            # Clean old entries (> 1 hour, > 1 day)
            self._clean_old_entries(user_id, now)

            # Get user-specific limits or defaults
            max_per_hour, max_per_day = self._user_limits.get(
                user_id, (self.max_per_hour, self.max_per_day)
            )

            # Check limits using user-specific values
            if len(self._hourly_counts[user_id]) >= max_per_hour:
                return False
            if len(self._daily_counts[user_id]) >= max_per_day:
                return False

            # Record this notification
            self._hourly_counts[user_id].append(now)
            self._daily_counts[user_id].append(now)

            return True

Shared Singleton Pattern with Per-User Configuration:

class NotificationManager:
    _shared_rate_limiter: Optional["RateLimiter"] = None
    _rate_limiter_lock = threading.Lock()

    def __init__(self, settings_snapshot: Dict[str, Any], user_id: str):
        with NotificationManager._rate_limiter_lock:
            if NotificationManager._shared_rate_limiter is None:
                # Create shared rate limiter with defaults
                NotificationManager._shared_rate_limiter = RateLimiter(
                    max_per_hour=settings_snapshot.get("notifications.rate_limit_per_hour", 10),
                    max_per_day=settings_snapshot.get("notifications.rate_limit_per_day", 50),
                )

            self._rate_limiter = NotificationManager._shared_rate_limiter

            # Configure per-user limits (user_id is required)
            max_per_hour = settings_snapshot.get("notifications.rate_limit_per_hour", 10)
            max_per_day = settings_snapshot.get("notifications.rate_limit_per_day", 50)
            self._rate_limiter.set_user_limits(user_id, max_per_hour, max_per_day)

Key Points:

• One rate limiter instance across all NotificationManager instances (singleton)
• Per-user rate limits: Each user can have different limits based on their settings
• Per-user counters: Each user has independent notification counters
• User isolation: One user hitting their limit does NOT affect others
• Thread-safe with threading.Lock() for all operations
• Automatic cleanup of old entries (> 1 hour, > 1 day)
• Periodic cleanup of inactive users (every 24 hours)
• Memory efficient: ~24 bytes per user for limit storage

Configuration

Required Settings

# Service URL (required) - comma-separated list
notifications.service_url = "discord://webhook_id/token,mailto://user:pass@smtp.gmail.com"

# Event-specific toggles (default: False for most events)
notifications.on_research_completed = True   # Default: True
notifications.on_research_failed = True      # Default: True
notifications.on_research_queued = False     # Default: False
notifications.on_subscription_update = True  # Default: True
notifications.on_subscription_error = False  # Default: False
notifications.on_api_quota_warning = False   # Default: False
notifications.on_auth_issue = False          # Default: False

# Rate limits (per-user, configured independently for each user)
notifications.rate_limit_per_hour = 10   # Max notifications per hour (per user)
notifications.rate_limit_per_day = 50    # Max notifications per day (per user)

# URL configuration (for clickable links)
app.external_url = "https://ldr.example.com"  # Preferred
# OR
app.host = "localhost"
app.port = 5000

Testing Notifications

from local_deep_research.notifications.manager import NotificationManager

# Create manager with settings and required user_id
notification_manager = NotificationManager(
    settings_snapshot={},
    user_id="test_user"  # Required for per-user rate limiting
)

# Test a service URL
result = notification_manager.test_service("discord://webhook_id/webhook_token")
print(result)  # {'success': True, 'message': 'Test notification sent successfully'}

Error Handling

Rate Limit Exceeded

try:
    notification_manager.send_notification(
        event_type=EventType.RESEARCH_COMPLETED,
        context=context,
    )
except RateLimitError as e:
    logger.warning(f"Rate limit exceeded: {e}")
    # Notification not sent, user needs to wait
    # Note: This only affects the user that the manager was created for

Send Failure (After 3 Retries)

try:
    result = service.send(title="...", body="...", service_urls="...")
except SendError as e:
    logger.error(f"Failed to send notification after 3 attempts: {e}")
    # All retry attempts exhausted

No Service URLs Configured

result = notification_manager.send_notification(...)
# Returns: False (no error raised)
# Log: "No notification service URLs configured for user {user_id}"

Notifications Disabled for Event Type

# settings_snapshot = {"notifications.on_research_completed": False}
result = notification_manager.send_notification(
    event_type=EventType.RESEARCH_COMPLETED,
    context=context
)
# Returns: False (respects user preference)
# Log: "Notifications disabled for event type: research_completed"

Architecture Diagram

[Research Thread]
      ↓
cleanup_research_resources()
      ↓
queue_processor.notify_research_completed(username, research_id, password)
      ↓
[Queue Processor - Main Thread]
      ↓
get_user_db_session(username, password) → [Encrypted DB]
      ↓
SettingsManager(session).get_settings_snapshot() → settings_snapshot
      ↓
ResearchHistory.query.filter_by(id=research_id).first() → research details
      ↓
build_notification_url() → full_url
      ↓
NotificationManager(settings_snapshot=settings_snapshot, user_id=username)
      ↓
notification_manager.send_notification(
    event_type=EventType.RESEARCH_COMPLETED,
    context={query, research_id, summary, url}
)
      ↓
[NotificationManager]
      ├─ Check: notifications.on_research_completed
      ├─ Check: Per-user rate limiter (hourly/daily)
      └─ Get: notifications.service_url
      ↓
NotificationService.send_event(event_type, context, service_urls)
      ↓
NotificationTemplate.format(event_type, context) → {title, body}
      ↓
NotificationService.send(title, body, service_urls)
      ↓
[Retry Loop: 3 attempts with exponential backoff]
      ├─ Attempt 1: Immediate
      ├─ Attempt 2: Wait 0.5s
      └─ Attempt 3: Wait 1.0s
      ↓
Apprise.add(service_urls)
      ↓
Apprise.notify(title=title, body=body)
      ↓
[Apprise - Delivery]
      ├─ Discord webhook
      ├─ SMTP email
      └─ Other services...

Key Design Decisions

1. Settings Snapshot Pattern: Avoids thread-safety issues with database sessions
2. Shared Rate Limiter with Per-User Limits: Single rate limiter instance ensures correct per-user enforcement while maintaining separate limits and counters for each user
3. User Isolation: Each user's rate limits are independent - one user hitting their limit does not affect others
4. Temporary Apprise Instances: Created per-send and automatically garbage collected
5. Exponential Backoff: 3 retry attempts with increasing delays (0.5s → 1.0s → 2.0s)
6. Encrypted Storage: Service URLs stored encrypted in per-user SQLCipher database
7. URL Masking: Credentials hidden in logs (e.g., discord://webhook_id/***)
8. No Session in Manager: NotificationManager never receives database session (thread-safe)
9. Error Sanitization: Error messages sanitized in notifications to prevent information exposure

Additional Notification Events

Research Queued Notifications

Sent when research is added to the queue:

• Event: EventType.RESEARCH_QUEUED
• Triggered from: web/queue/manager.py when adding research to queue
• Context: query, position, wait_time
• Default: Disabled (opt-in)

API Quota Warnings

Sent when API rate limits are exceeded:

• Event: EventType.API_QUOTA_WARNING
• Triggered from: error_handling/error_reporter.py when detecting rate limit errors
• Context: service, current, limit, reset_time
• Default: Disabled (opt-in)

Authentication Issues

Sent when API authentication fails:

• Event: EventType.AUTH_ISSUE
• Triggered from: error_handling/error_reporter.py when detecting auth errors
• Context: service
• Default: Disabled (opt-in)

Subscription Notifications

Sent when subscriptions update or fail:

• Events: EventType.SUBSCRIPTION_UPDATE, EventType.SUBSCRIPTION_ERROR
• Context: subscription_name, subscription_id, item_count/error, url

Testing

Run notification tests:

pdm run python -m pytest tests/notifications/ -v

All 98 tests passing ✓ (including 7 new per-user rate limiting tests)