613 Notification Flow 5dc95fbc

# 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:

```python
# 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:

```python
# 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`)

```python
# 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:

```python
# 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`)

```python
# 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`)

```python
# 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`):**
```python
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`)

```python
# 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:

```python
# 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:**
```python
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:**

```python
# 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

```python
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:**
```python
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

```python
# 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

```python
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

```python
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)

```python
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

```python
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

```python
# 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:
```bash
pdm run python -m pytest tests/notifications/ -v
```

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