diff --git a/CHANGELOG.md b/CHANGELOG.md index 5b96c68..83ccf31 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] ### Added +- **Apprise alerting**: New `NotificationService` using [Apprise](https://github.com/caronc/apprise) for multi-channel push notifications (Telegram, Slack, Discord, webhooks, and 80+ other services via a single URL scheme). + - `send_user_notification` — sends to all enabled per-user Apprise channels on processing errors or failures. + - `send_admin_notification` — sends to all enabled admin-wide channels for system events. + - `test_notification` — validates an Apprise URL by dispatching a test message. +- **`NotificationConfig` model**: Added `name` (friendly label) and `apprise_url` (nullable Apprise URL) columns. +- **`AdminNotificationConfig` model**: New table (`admin_notification_configs`) for system-wide admin alert channels with `name`, `apprise_url`, `is_enabled`, `notify_on_errors`, `notify_on_system_events`, and `description` fields. +- **Notifications API** (`/api/v1/notifications`): Full CRUD endpoints (GET/POST/PUT/DELETE) plus a `/test` endpoint for user notification configs. +- **Admin Notifications API** (`/api/v1/admin/notifications`): Full CRUD + `/test` endpoints for admin notification configs, superuser-only. +- **Task integration**: `process_mail_account` now calls `send_user_notification` on Gmail credential revocation, per-email forwarding failures, and unhandled processing exceptions. + +### Changed +- `NotificationConfigBase` schema: `name` is now a required field; `apprise_url` is an optional field; `config` (channel-specific JSON) is now optional with a default of `{}` (previously required). Existing clients must be updated to supply `name`. + + - **Prometheus metrics** (`/metrics` endpoint on the FastAPI backend, scraped every 15 s): - **HTTP layer** — `http_requests_total` (counter, labelled `method`/`endpoint`/`status_code`) and `http_request_duration_seconds` (histogram). Path segments that are numeric IDs are normalised to `{id}` to avoid label-set explosion. - **Mail processing** — `mail_processing_runs_total` (counter, by `status`: `completed` / `partial_failure` / `failed`), `mail_processing_emails_total` (counter, by `operation`: `fetched` / `forwarded` / `failed`), `mail_processing_duration_seconds` (histogram), `active_mail_accounts_total` (gauge — set each scheduler cycle). diff --git a/backend/app/api/v1/endpoints/admin.py b/backend/app/api/v1/endpoints/admin.py index ddab0e9..08a4463 100644 --- a/backend/app/api/v1/endpoints/admin.py +++ b/backend/app/api/v1/endpoints/admin.py @@ -13,6 +13,7 @@ from app.models.database_models import ( ProcessingRun, SubscriptionPlan, SubscriptionTier, + AdminNotificationConfig, ) from app.models.schemas import ( AdminUserListResponse, @@ -21,7 +22,13 @@ from app.models.schemas import ( SubscriptionPlanResponse, SubscriptionPlanCreate, SubscriptionPlanUpdate, + AdminNotificationConfigCreate, + AdminNotificationConfigUpdate, + AdminNotificationConfigResponse, + NotificationTestRequest, + NotificationTestResponse, ) +from app.services.notification_service import test_notification router = APIRouter() @@ -282,3 +289,114 @@ async def delete_plan( ) await db.delete(plan) await db.commit() + + +# ── Admin notification config management ────────────────────────────────────── + + +@router.get("/notifications", response_model=List[AdminNotificationConfigResponse]) +async def list_admin_notification_configs( + current_user: User = Depends(get_current_superuser), + db: AsyncSession = Depends(get_db), +): + """List all admin notification configurations (admin only)""" + result = await db.execute(select(AdminNotificationConfig)) + return result.scalars().all() + + +@router.post( + "/notifications", + response_model=AdminNotificationConfigResponse, + status_code=status.HTTP_201_CREATED, +) +async def create_admin_notification_config( + config_in: AdminNotificationConfigCreate, + current_user: User = Depends(get_current_superuser), + db: AsyncSession = Depends(get_db), +): + """Create a new admin notification configuration (admin only)""" + config = AdminNotificationConfig(**config_in.dict()) + db.add(config) + await db.commit() + await db.refresh(config) + return config + + +@router.get( + "/notifications/{config_id}", response_model=AdminNotificationConfigResponse +) +async def get_admin_notification_config( + config_id: int, + current_user: User = Depends(get_current_superuser), + db: AsyncSession = Depends(get_db), +): + """Get a specific admin notification configuration (admin only)""" + result = await db.execute( + select(AdminNotificationConfig).where(AdminNotificationConfig.id == config_id) + ) + config = result.scalar_one_or_none() + if not config: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Admin notification config not found", + ) + return config + + +@router.put( + "/notifications/{config_id}", response_model=AdminNotificationConfigResponse +) +async def update_admin_notification_config( + config_id: int, + config_in: AdminNotificationConfigUpdate, + current_user: User = Depends(get_current_superuser), + db: AsyncSession = Depends(get_db), +): + """Update an admin notification configuration (admin only)""" + result = await db.execute( + select(AdminNotificationConfig).where(AdminNotificationConfig.id == config_id) + ) + config = result.scalar_one_or_none() + if not config: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Admin notification config not found", + ) + + update_data = config_in.dict(exclude_unset=True) + for field, value in update_data.items(): + setattr(config, field, value) + + await db.commit() + await db.refresh(config) + return config + + +@router.delete("/notifications/{config_id}", status_code=status.HTTP_204_NO_CONTENT) +async def delete_admin_notification_config( + config_id: int, + current_user: User = Depends(get_current_superuser), + db: AsyncSession = Depends(get_db), +): + """Delete an admin notification configuration (admin only)""" + result = await db.execute( + select(AdminNotificationConfig).where(AdminNotificationConfig.id == config_id) + ) + config = result.scalar_one_or_none() + if not config: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Admin notification config not found", + ) + await db.delete(config) + await db.commit() + + +@router.post("/notifications/test", response_model=NotificationTestResponse) +async def test_admin_notification_config( + request: NotificationTestRequest, + current_user: User = Depends(get_current_superuser), +): + """Test an admin notification channel by sending a test message (admin only)""" + success, message = await test_notification(request.apprise_url) + return NotificationTestResponse(success=success, message=message) diff --git a/backend/app/api/v1/endpoints/notifications.py b/backend/app/api/v1/endpoints/notifications.py index 77b5480..4706028 100644 --- a/backend/app/api/v1/endpoints/notifications.py +++ b/backend/app/api/v1/endpoints/notifications.py @@ -1,7 +1,7 @@ """Notification configuration endpoints""" from typing import List -from fastapi import APIRouter, Depends, status +from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select @@ -10,8 +10,12 @@ from app.core.deps import get_current_active_user from app.models.database_models import User, NotificationConfig from app.models.schemas import ( NotificationConfigCreate, + NotificationConfigUpdate, NotificationConfigResponse, + NotificationTestRequest, + NotificationTestResponse, ) +from app.services.notification_service import test_notification router = APIRouter() @@ -24,7 +28,7 @@ async def create_notification_config( current_user: User = Depends(get_current_active_user), db: AsyncSession = Depends(get_db), ): - """Create notification configuration""" + """Create a new notification configuration""" config = NotificationConfig(user_id=current_user.id, **config_in.dict()) db.add(config) await db.commit() @@ -37,8 +41,94 @@ async def list_notification_configs( current_user: User = Depends(get_current_active_user), db: AsyncSession = Depends(get_db), ): - """List all notification configurations""" + """List all notification configurations for the current user""" result = await db.execute( select(NotificationConfig).where(NotificationConfig.user_id == current_user.id) ) return result.scalars().all() + + +@router.get("/{config_id}", response_model=NotificationConfigResponse) +async def get_notification_config( + config_id: int, + current_user: User = Depends(get_current_active_user), + db: AsyncSession = Depends(get_db), +): + """Get a specific notification configuration""" + result = await db.execute( + select(NotificationConfig).where( + NotificationConfig.id == config_id, + NotificationConfig.user_id == current_user.id, + ) + ) + config = result.scalar_one_or_none() + if not config: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Notification config not found", + ) + return config + + +@router.put("/{config_id}", response_model=NotificationConfigResponse) +async def update_notification_config( + config_id: int, + config_in: NotificationConfigUpdate, + current_user: User = Depends(get_current_active_user), + db: AsyncSession = Depends(get_db), +): + """Update a notification configuration""" + result = await db.execute( + select(NotificationConfig).where( + NotificationConfig.id == config_id, + NotificationConfig.user_id == current_user.id, + ) + ) + config = result.scalar_one_or_none() + if not config: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Notification config not found", + ) + + update_data = config_in.dict(exclude_unset=True) + for field, value in update_data.items(): + setattr(config, field, value) + + await db.commit() + await db.refresh(config) + return config + + +@router.delete("/{config_id}", status_code=status.HTTP_204_NO_CONTENT) +async def delete_notification_config( + config_id: int, + current_user: User = Depends(get_current_active_user), + db: AsyncSession = Depends(get_db), +): + """Delete a notification configuration""" + result = await db.execute( + select(NotificationConfig).where( + NotificationConfig.id == config_id, + NotificationConfig.user_id == current_user.id, + ) + ) + config = result.scalar_one_or_none() + if not config: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Notification config not found", + ) + + await db.delete(config) + await db.commit() + + +@router.post("/test", response_model=NotificationTestResponse) +async def test_notification_config( + request: NotificationTestRequest, + current_user: User = Depends(get_current_active_user), +): + """Test a notification channel by sending a test message""" + success, message = await test_notification(request.apprise_url) + return NotificationTestResponse(success=success, message=message) diff --git a/backend/app/models/database_models.py b/backend/app/models/database_models.py index 7932792..0e04309 100644 --- a/backend/app/models/database_models.py +++ b/backend/app/models/database_models.py @@ -289,6 +289,11 @@ class NotificationConfig(Base): Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=False ) + name = Column(String(255), nullable=False, default="My Notification") + apprise_url = Column( + Text, nullable=True + ) # The Apprise URL e.g. tgram://token/chatid + # Channel details channel: Column[str] = Column(SQLEnum(NotificationChannel), nullable=False) is_enabled = Column(Boolean, default=True) @@ -590,3 +595,29 @@ class AppSetting(Base): onupdate=lambda: datetime.now(timezone.utc), nullable=False, ) + + +class AdminNotificationConfig(Base): + """System-wide admin notification channels""" + + __tablename__ = "admin_notification_configs" + + id = Column(Integer, primary_key=True, index=True) + name = Column(String(255), nullable=False) + apprise_url = Column(Text, nullable=False) + is_enabled = Column(Boolean, default=True) + notify_on_errors = Column(Boolean, default=True) + notify_on_system_events = Column(Boolean, default=True) + description = Column(Text, nullable=True) + + created_at = Column( + DateTime(timezone=True), + default=lambda: datetime.now(timezone.utc), + nullable=False, + ) + updated_at = Column( + DateTime(timezone=True), + default=lambda: datetime.now(timezone.utc), + onupdate=lambda: datetime.now(timezone.utc), + nullable=False, + ) diff --git a/backend/app/models/schemas.py b/backend/app/models/schemas.py index c5882ab..6ec3f1b 100644 --- a/backend/app/models/schemas.py +++ b/backend/app/models/schemas.py @@ -219,9 +219,13 @@ class ProcessingLogResponse(BaseModel): # Notification Config Schemas class NotificationConfigBase(BaseModel): + name: str = Field( + ..., max_length=255, description="Friendly name for this notification channel" + ) channel: NotificationChannel + apprise_url: Optional[str] = Field(None, description="Apprise notification URL") is_enabled: bool = True - config: Dict[str, Any] + config: Dict[str, Any] = Field(default_factory=dict) notify_on_errors: bool = True notify_on_success: bool = False notify_threshold: int = Field(default=3, gt=0, le=100) @@ -232,6 +236,9 @@ class NotificationConfigCreate(NotificationConfigBase): class NotificationConfigUpdate(BaseModel): + name: Optional[str] = Field(None, max_length=255) + channel: Optional[NotificationChannel] = None + apprise_url: Optional[str] = None is_enabled: Optional[bool] = None config: Optional[Dict[str, Any]] = None notify_on_errors: Optional[bool] = None @@ -248,6 +255,46 @@ class NotificationConfigResponse(NotificationConfigBase): model_config = ConfigDict(from_attributes=True) +class NotificationTestRequest(BaseModel): + apprise_url: str = Field(..., description="Apprise URL to test") + + +class NotificationTestResponse(BaseModel): + success: bool + message: str + + +# Admin Notification Config Schemas +class AdminNotificationConfigBase(BaseModel): + name: str = Field(..., max_length=255) + apprise_url: str = Field(..., description="Apprise notification URL") + is_enabled: bool = True + notify_on_errors: bool = True + notify_on_system_events: bool = True + description: Optional[str] = None + + +class AdminNotificationConfigCreate(AdminNotificationConfigBase): + pass + + +class AdminNotificationConfigUpdate(BaseModel): + name: Optional[str] = Field(None, max_length=255) + apprise_url: Optional[str] = None + is_enabled: Optional[bool] = None + notify_on_errors: Optional[bool] = None + notify_on_system_events: Optional[bool] = None + description: Optional[str] = None + + +class AdminNotificationConfigResponse(AdminNotificationConfigBase): + id: int + created_at: datetime + updated_at: datetime + + model_config = ConfigDict(from_attributes=True) + + # Subscription Schemas class SubscriptionPlanResponse(BaseModel): id: int diff --git a/backend/app/services/notification_service.py b/backend/app/services/notification_service.py new file mode 100644 index 0000000..c017e53 --- /dev/null +++ b/backend/app/services/notification_service.py @@ -0,0 +1,143 @@ +""" +Notification service using Apprise for multi-channel alerting. + +Supports: +- User-specific notifications (per-user Apprise URLs) +- Admin-wide system notifications (system-level alerts) +- Test notifications to verify configuration +""" + +import logging + +import apprise + +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy import select + +from app.models.database_models import NotificationConfig, AdminNotificationConfig + +logger = logging.getLogger(__name__) + + +async def send_user_notification( + db: AsyncSession, + user_id: int, + title: str, + body: str, + notify_on_error: bool = True, +) -> int: + """ + Send a notification to all enabled notification channels for a given user. + + Args: + db: Database session + user_id: The user to notify + title: Notification title/subject + body: Notification body text + notify_on_error: If True, only sends to channels with notify_on_errors=True + If False, only sends to channels with notify_on_success=True + + Returns: + Number of channels notified successfully + """ + result = await db.execute( + select(NotificationConfig).where( + NotificationConfig.user_id == user_id, + NotificationConfig.is_enabled == True, # noqa: E712 + NotificationConfig.apprise_url.isnot(None), + ) + ) + configs = result.scalars().all() + + if not configs: + return 0 + + sent = 0 + for config in configs: + if notify_on_error and not config.notify_on_errors: + continue + if not notify_on_error and not config.notify_on_success: + continue + + try: + success = await _send_apprise(str(config.apprise_url), title, body) + if success: + sent += 1 + except Exception as exc: + logger.warning( + "Failed to send notification via channel %s (user %s): %s", + config.id, + user_id, + exc, + ) + + return sent + + +async def send_admin_notification( + db: AsyncSession, + title: str, + body: str, +) -> int: + """ + Send a notification to all enabled admin notification channels. + + Returns: + Number of channels notified successfully + """ + result = await db.execute( + select(AdminNotificationConfig).where( + AdminNotificationConfig.is_enabled == True, # noqa: E712 + AdminNotificationConfig.notify_on_errors == True, # noqa: E712 + ) + ) + configs = result.scalars().all() + + if not configs: + return 0 + + sent = 0 + for config in configs: + try: + success = await _send_apprise(str(config.apprise_url), title, body) + if success: + sent += 1 + except Exception as exc: + logger.warning( + "Failed to send admin notification via channel %s: %s", + config.id, + exc, + ) + + return sent + + +async def test_notification(apprise_url: str) -> tuple[bool, str]: + """ + Send a test notification to the given Apprise URL. + + Returns: + (success, message) tuple + """ + try: + success = await _send_apprise( + apprise_url, + title="InboxRescue: Test Notification", + body="This is a test notification from InboxRescue. Your notification channel is configured correctly!", + ) + if success: + return True, "Test notification sent successfully" + return False, "Notification delivery failed (check your Apprise URL)" + except Exception as exc: + return False, f"Error sending test notification: {exc}" + + +async def _send_apprise(url: str, title: str, body: str) -> bool: + """Internal helper – create an Apprise instance, load the URL, and notify.""" + ap = apprise.Apprise() + if not ap.add(url): + logger.warning("Apprise could not parse URL: %s", url[:60]) + return False + + result = await ap.async_notify(title=title, body=body) + return bool(result) diff --git a/backend/app/workers/tasks.py b/backend/app/workers/tasks.py index 30287a1..ccb42f3 100644 --- a/backend/app/workers/tasks.py +++ b/backend/app/workers/tasks.py @@ -33,6 +33,7 @@ from app.models.database_models import ( from app.services.mail_processor import MailProcessor from app.services.gmail_service import GmailService from app.services.config_service import ConfigService +from app.services.notification_service import send_user_notification from app.core.config import settings from sqlalchemy import select, delete @@ -239,6 +240,18 @@ async def process_mail_account(account_id: int): f"Gmail credentials revoked for user {account.user_id}. " "User must re-authorise." ) + try: + await send_user_notification( + db=db, + user_id=account.user_id, + title="InboxRescue: Gmail Authorization Expired", + body=f"Your Gmail credentials for account '{account.name}' have been revoked. Please re-authorize Gmail access in Settings.", + notify_on_error=True, + ) + except Exception as notify_exc: + logger.warning( + f"Failed to send revocation notification: {notify_exc}" + ) logger.error(f"Error delivering email: {e}") emails_failed += 1 @@ -286,6 +299,16 @@ async def process_mail_account(account_id: int): account.status = AccountStatus.ERROR # type: ignore[assignment] account.last_error_at = datetime.now(timezone.utc) # type: ignore[assignment] account.last_error_message = f"{emails_failed} emails failed to forward" # type: ignore[assignment] + try: + await send_user_notification( + db=db, + user_id=account.user_id, + title="InboxRescue: Mail Forwarding Failures", + body=f"Mail account '{account.name}': {emails_failed} email(s) failed to forward.", + notify_on_error=True, + ) + except Exception as notify_exc: + logger.warning(f"Failed to send notification: {notify_exc}") await db.commit() @@ -338,6 +361,20 @@ async def process_mail_account(account_id: int): account.last_error_at = datetime.now(timezone.utc) # type: ignore[assignment] account.last_error_message = str(e) # type: ignore[assignment] + # Notify user about the error + try: + await send_user_notification( + db=db, + user_id=account.user_id, + title="InboxRescue: Mail Processing Error", + body=f"Error processing mail account '{account.name}': {e}", + notify_on_error=True, + ) + except Exception as notify_exc: + logger.warning( + f"Failed to send error notification: {notify_exc}" + ) + await db.commit()