bambuddy/backend/app/api/routes/notifications.py

"""API routes for notification providers."""

import json
import logging
import time
from collections import defaultdict, deque
from datetime import datetime, timedelta, timezone

from fastapi import APIRouter, Depends, HTTPException, Query
from fastapi.responses import FileResponse
from sqlalchemy import delete, desc, func, select, update
from sqlalchemy.ext.asyncio import AsyncSession

from backend.app.core.auth import (
    RequestPrinterScope,
    RequirePermissionIfAuthEnabled,
    ScopedCaller,
    probe_permissions_if_auth_enabled,
    require_notification_send,
)
from backend.app.core.database import get_db
from backend.app.core.permissions import Permission
from backend.app.core.printer_scope import PrinterScope, ensure_printers_exist
from backend.app.models.notification import NotificationLog, NotificationProvider
from backend.app.models.notification_lock_screen_widget import NotificationLockScreenWidget
from backend.app.models.user import User
from backend.app.schemas.notification import (
    AppMessage,
    AppMessageChannel,
    AppMessageResult,
    NotificationLogResponse,
    NotificationLogStats,
    NotificationProviderCreate,
    NotificationProviderResponse,
    NotificationProviderUpdate,
    NotificationTestRequest,
    NotificationTestResponse,
)
from backend.app.services.notification_service import notification_service
from backend.app.services.notify_client import NotifyError, notify_credentials
from backend.app.services.notify_live_activities import notify_live_activities
from backend.app.services.notify_widgets import notify_widgets
from backend.app.services.secret_fields import (
    SecretConfigError,
    mask_config,
    merge_config,
    reject_placeholder,
)
from backend.app.services.telegram_reactions import telegram_reaction_poller
from backend.app.utils.notification_photos import find_notification_photo

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/notifications", tags=["notifications"])


def _notify_providers_changed() -> None:
    """Wake optional workers after a provider edit, without waiting for their HTTP work."""
    for worker in (notify_live_activities, notify_widgets):
        try:
            worker.providers_changed()
        except Exception:
            logger.exception("Could not refresh Notify worker configuration")


async def _resync_reaction_poller():
    """Start/stop Telegram reaction polls after a provider changed (#3046).

    Never fails the request: the provider row is already saved, and the
    poller catches up on the next restart at worst.
    """
    try:
        await telegram_reaction_poller.sync()
    except Exception as e:
        logger.warning("Telegram reaction poller resync failed: %s", e)


async def delete_provider(db: AsyncSession, provider: NotificationProvider) -> bool:
    """Delete a provider (the caller commits, then calls after_providers_deleted).

    A Bambuddy Notify provider is first retired and its live activities and
    widgets queued for remote cleanup, while the rows that name them still
    exist. Returns whether it was one.
    """
    notify = provider.provider_type == "notify"
    if notify:
        config = json.loads(provider.config) if isinstance(provider.config, str) else provider.config
        # Retire this provider and capture owned IDs before FK cascade removes
        # them. Queue remote cleanup; HTTP must never delay this request.
        provider.enabled = False
        await db.commit()
        await notify_live_activities.schedule_cleanup(provider.id, config)
        await notify_widgets.schedule_cleanup(provider.id, config)
    await db.delete(provider)
    return notify


async def after_providers_deleted(*, notify: bool) -> None:
    """What deleted providers leave to do once the delete is committed."""
    if notify:
        _notify_providers_changed()
    await _resync_reaction_poller()


def _stored_config(provider: NotificationProvider) -> dict:
    return json.loads(provider.config) if isinstance(provider.config, str) else provider.config


def _provider_to_dict(provider: NotificationProvider) -> dict:
    """Convert a NotificationProvider model to a response dictionary.

    Secrets are write-only: a set one is shown as the placeholder.
    """
    return {
        "id": provider.id,
        "name": provider.name,
        "provider_type": provider.provider_type,
        "enabled": provider.enabled,
        "config": mask_config(_stored_config(provider)),
        "attach_photo": provider.attach_photo,
        # Print lifecycle events
        "on_print_start": provider.on_print_start,
        "on_print_complete": provider.on_print_complete,
        "on_print_failed": provider.on_print_failed,
        "on_print_stopped": provider.on_print_stopped,
        "on_print_progress": provider.on_print_progress,
        "on_print_missing_spool_assignment": provider.on_print_missing_spool_assignment,
        "on_billing_charge_failed": provider.on_billing_charge_failed,
        # Printer status events
        "on_printer_offline": provider.on_printer_offline,
        "on_printer_error": provider.on_printer_error,
        "on_ai_failure_detection": provider.on_ai_failure_detection,
        "on_filament_low": provider.on_filament_low,
        "on_maintenance_due": provider.on_maintenance_due,
        # Home Assistant sensor alerts (#1148, #2824). Both directions of this
        # file are hand-maintained field maps, so a column missing here reads
        # back as the schema default (False) no matter what the row holds.
        "on_ha_sensor_alert": provider.on_ha_sensor_alert,
        "on_location_ha_sensor_alert": provider.on_location_ha_sensor_alert,
        # AMS environmental alarms (regular AMS)
        "on_ams_humidity_high": provider.on_ams_humidity_high,
        "on_ams_temperature_high": provider.on_ams_temperature_high,
        "on_ams_drying_suspended": provider.on_ams_drying_suspended,
        # AMS-HT environmental alarms
        "on_ams_ht_humidity_high": provider.on_ams_ht_humidity_high,
        "on_ams_ht_temperature_high": provider.on_ams_ht_temperature_high,
        # Build plate detection
        "on_plate_not_empty": provider.on_plate_not_empty,
        "on_plate_clear_required": provider.on_plate_clear_required,
        # Post-print outcome confirmation (#1898)
        "on_print_confirm_request": provider.on_print_confirm_request,
        # Rows from before #3046 hold NULL here; "buttons" is what they did.
        "telegram_verdict_mode": provider.telegram_verdict_mode or "buttons",
        # Bed cooled
        "on_bed_cooled": provider.on_bed_cooled,
        # First layer complete
        "on_first_layer_complete": provider.on_first_layer_complete,
        # Messages from connected apps
        "on_app_message": bool(provider.on_app_message),
        # Inventory stock alerts. Absent here, the toggles above always read
        # back off no matter what the row holds — the same hand-maintained
        # field map the Home Assistant comment warns about.
        "on_stock_reorder_alert": provider.on_stock_reorder_alert,
        "on_stock_break_alert": provider.on_stock_break_alert,
        # Print queue events
        "on_queue_job_added": provider.on_queue_job_added,
        "on_queue_job_assigned": provider.on_queue_job_assigned,
        "on_queue_job_started": provider.on_queue_job_started,
        "on_queue_job_waiting": provider.on_queue_job_waiting,
        "on_queue_job_skipped": provider.on_queue_job_skipped,
        "on_queue_job_failed": provider.on_queue_job_failed,
        "on_queue_completed": provider.on_queue_completed,
        # Quiet hours
        "quiet_hours_enabled": provider.quiet_hours_enabled,
        "quiet_hours_start": provider.quiet_hours_start,
        "quiet_hours_end": provider.quiet_hours_end,
        # Daily digest
        "daily_digest_enabled": provider.daily_digest_enabled,
        "daily_digest_time": provider.daily_digest_time,
        # Printer filter
        "printer_id": provider.printer_id,
        # Status tracking
        "last_success": provider.last_success,
        "last_error": provider.last_error,
        "last_error_at": provider.last_error_at,
        # Timestamps
        "created_at": provider.created_at,
        "updated_at": provider.updated_at,
    }


# ============================================================================
# Provider List/Create Routes (no path parameters)
# ============================================================================


def _provider_in_scope(provider: NotificationProvider, printer_scope: PrinterScope) -> bool:
    """Whether a caller limited to some printers may see and manage ``provider``.

    Only one bound to one of their printers: a provider for all printers (or
    another printer) delivers events of printers outside their groups.
    """
    if printer_scope.is_unrestricted:
        return True
    return provider.printer_id is not None and printer_scope.allows(provider.printer_id)


def _log_scope_clause(printer_scope: PrinterScope):
    """The log entries a caller limited to some printers sees (None: all).

    Those of their printers, and those with no printer sent by a provider they
    can see: an entry with no printer belongs to whoever owns its provider.
    """
    if printer_scope.printer_ids is None:
        return None
    own_providers = select(NotificationProvider.id).where(
        NotificationProvider.printer_id.in_(printer_scope.printer_ids)
    )
    return NotificationLog.printer_id.in_(printer_scope.printer_ids) | (
        NotificationLog.printer_id.is_(None) & NotificationLog.provider_id.in_(own_providers)
    )


async def _load_provider(db: AsyncSession, provider_id: int, printer_scope: PrinterScope) -> NotificationProvider:
    provider = await db.get(NotificationProvider, provider_id)
    if provider is None or not _provider_in_scope(provider, printer_scope):
        raise HTTPException(status_code=404, detail="Notification provider not found")
    return provider


def _ensure_target_in_scope(printer_id: int | None, printer_scope: PrinterScope) -> None:
    """A caller limited to some printers binds a provider to one of them, never to all."""
    if printer_scope.is_unrestricted:
        return
    if printer_id is None:
        raise HTTPException(status_code=400, detail="Choose one of your printers for this provider")
    printer_scope.ensure(printer_id)


@router.get("/", response_model=list[NotificationProviderResponse])
async def list_notification_providers(
    db: AsyncSession = Depends(get_db),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.NOTIFICATIONS_READ),
    printer_scope: PrinterScope = RequestPrinterScope,
):
    """List all notification providers."""
    result = await db.execute(select(NotificationProvider).order_by(NotificationProvider.created_at.desc()))
    providers = [p for p in result.scalars().all() if _provider_in_scope(p, printer_scope)]

    return [_provider_to_dict(provider) for provider in providers]


@router.post("/", response_model=NotificationProviderResponse)
async def create_notification_provider(
    provider_data: NotificationProviderCreate,
    db: AsyncSession = Depends(get_db),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.NOTIFICATIONS_CREATE),
    printer_scope: PrinterScope = RequestPrinterScope,
):
    """Create a new notification provider."""
    _ensure_target_in_scope(provider_data.printer_id, printer_scope)
    await ensure_printers_exist(db, [provider_data.printer_id])
    try:
        reject_placeholder(provider_data.config)
    except SecretConfigError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from None
    if provider_data.provider_type.value == "notify":
        try:
            notify_credentials(provider_data.config)
        except NotifyError as exc:
            raise HTTPException(status_code=422, detail=str(exc)) from None
    provider = NotificationProvider(
        name=provider_data.name,
        provider_type=provider_data.provider_type.value,
        enabled=provider_data.enabled,
        config=json.dumps(provider_data.config),
        attach_photo=provider_data.attach_photo,
        # Print lifecycle events
        on_print_start=provider_data.on_print_start,
        on_print_complete=provider_data.on_print_complete,
        on_print_failed=provider_data.on_print_failed,
        on_print_stopped=provider_data.on_print_stopped,
        on_print_progress=provider_data.on_print_progress,
        on_print_missing_spool_assignment=provider_data.on_print_missing_spool_assignment,
        on_billing_charge_failed=provider_data.on_billing_charge_failed,
        # Printer status events
        on_printer_offline=provider_data.on_printer_offline,
        on_printer_error=provider_data.on_printer_error,
        on_ai_failure_detection=provider_data.on_ai_failure_detection,
        on_filament_low=provider_data.on_filament_low,
        on_maintenance_due=provider_data.on_maintenance_due,
        # Home Assistant sensor alerts (#1148, #2824)
        on_ha_sensor_alert=provider_data.on_ha_sensor_alert,
        on_location_ha_sensor_alert=provider_data.on_location_ha_sensor_alert,
        # AMS environmental alarms (regular AMS)
        on_ams_humidity_high=provider_data.on_ams_humidity_high,
        on_ams_temperature_high=provider_data.on_ams_temperature_high,
        on_ams_drying_suspended=provider_data.on_ams_drying_suspended,
        # AMS-HT environmental alarms
        on_ams_ht_humidity_high=provider_data.on_ams_ht_humidity_high,
        on_ams_ht_temperature_high=provider_data.on_ams_ht_temperature_high,
        # Build plate detection
        on_plate_not_empty=provider_data.on_plate_not_empty,
        on_plate_clear_required=provider_data.on_plate_clear_required,
        # Post-print outcome confirmation (#1898)
        on_print_confirm_request=provider_data.on_print_confirm_request,
        telegram_verdict_mode=provider_data.telegram_verdict_mode,
        # Bed cooled
        on_bed_cooled=provider_data.on_bed_cooled,
        # First layer complete
        on_first_layer_complete=provider_data.on_first_layer_complete,
        on_app_message=provider_data.on_app_message,
        # Inventory stock alerts
        on_stock_reorder_alert=provider_data.on_stock_reorder_alert,
        on_stock_break_alert=provider_data.on_stock_break_alert,
        # Print queue events
        on_queue_job_added=provider_data.on_queue_job_added,
        on_queue_job_assigned=provider_data.on_queue_job_assigned,
        on_queue_job_started=provider_data.on_queue_job_started,
        on_queue_job_waiting=provider_data.on_queue_job_waiting,
        on_queue_job_skipped=provider_data.on_queue_job_skipped,
        on_queue_job_failed=provider_data.on_queue_job_failed,
        on_queue_completed=provider_data.on_queue_completed,
        # Quiet hours
        quiet_hours_enabled=provider_data.quiet_hours_enabled,
        quiet_hours_start=provider_data.quiet_hours_start,
        quiet_hours_end=provider_data.quiet_hours_end,
        # Daily digest
        daily_digest_enabled=provider_data.daily_digest_enabled,
        daily_digest_time=provider_data.daily_digest_time,
        # Printer filter
        printer_id=provider_data.printer_id,
    )

    db.add(provider)
    await db.commit()
    await db.refresh(provider)

    logger.info("Created notification provider: %s (%s)", provider.name, provider.provider_type)
    if provider.provider_type == "notify":
        _notify_providers_changed()
    await _resync_reaction_poller()

    return _provider_to_dict(provider)


# ============================================================================
# Static Path Routes (must come BEFORE parameterized routes)
# ============================================================================


@router.post("/test-config", response_model=NotificationTestResponse)
async def test_notification_config(
    test_request: NotificationTestRequest,
    db: AsyncSession = Depends(get_db),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.NOTIFICATIONS_CREATE),
    can_update: bool = Depends(probe_permissions_if_auth_enabled(Permission.NOTIFICATIONS_UPDATE)),
    printer_scope: PrinterScope = RequestPrinterScope,
):
    """Test notification configuration before saving.

    With provider_id (an edited provider), the stored secrets fill in the
    placeholders, for those who may edit providers.
    """
    config = test_request.config
    try:
        if test_request.provider_id is None:
            reject_placeholder(config)
        else:
            if not can_update:
                raise HTTPException(status_code=403, detail="Testing a saved provider requires notifications:update")
            provider = await _load_provider(db, test_request.provider_id, printer_scope)
            config = merge_config(
                provider.provider_type, _stored_config(provider), test_request.provider_type.value, config
            )
    except SecretConfigError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from None
    success, message = await notification_service.send_test_notification(
        test_request.provider_type.value, config, db, attach_photo=test_request.attach_photo
    )

    return NotificationTestResponse(success=success, message=message)


@router.post("/test-all")
async def test_all_notification_providers(
    db: AsyncSession = Depends(get_db),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.NOTIFICATIONS_UPDATE),
    printer_scope: PrinterScope = RequestPrinterScope,
):
    """Send a test notification to all enabled providers."""
    result = await db.execute(select(NotificationProvider).where(NotificationProvider.enabled.is_(True)))
    providers = [p for p in result.scalars().all() if _provider_in_scope(p, printer_scope)]

    if not providers:
        return {"tested": 0, "success": 0, "failed": 0, "results": []}

    results = []
    success_count = 0
    failed_count = 0

    for provider in providers:
        config = json.loads(provider.config) if isinstance(provider.config, str) else provider.config
        success, message = await notification_service.send_test_notification(
            provider.provider_type, config, db, attach_photo=provider.attach_photo
        )

        # Update provider status
        if success:
            provider.last_success = datetime.now(timezone.utc)
            success_count += 1
        else:
            provider.last_error = message
            provider.last_error_at = datetime.now(timezone.utc)
            failed_count += 1

        results.append(
            {
                "provider_id": provider.id,
                "provider_name": provider.name,
                "provider_type": provider.provider_type,
                "success": success,
                "message": message,
            }
        )

    await db.commit()

    return {
        "tested": len(providers),
        "success": success_count,
        "failed": failed_count,
        "results": results,
    }


# ============================================================================
# Notification Log Routes (must come BEFORE /{provider_id} routes)
# ============================================================================


@router.get("/logs", response_model=list[NotificationLogResponse])
async def get_notification_logs(
    limit: int = Query(default=100, ge=1, le=500),
    offset: int = Query(default=0, ge=0),
    provider_id: int | None = Query(default=None),
    event_type: str | None = Query(default=None),
    success: bool | None = Query(default=None),
    days: int | None = Query(default=7, ge=1, le=90, description="Filter logs from the last N days"),
    db: AsyncSession = Depends(get_db),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.NOTIFICATIONS_READ),
    printer_scope: PrinterScope = RequestPrinterScope,
):
    """Get notification logs with optional filters."""
    query = select(NotificationLog).order_by(desc(NotificationLog.created_at))
    if (scope_clause := _log_scope_clause(printer_scope)) is not None:
        query = query.where(scope_clause)

    # Apply filters
    if provider_id is not None:
        query = query.where(NotificationLog.provider_id == provider_id)
    if event_type is not None:
        query = query.where(NotificationLog.event_type == event_type)
    if success is not None:
        query = query.where(NotificationLog.success == success)
    if days is not None:
        cutoff = datetime.now(timezone.utc) - timedelta(days=days)
        query = query.where(NotificationLog.created_at >= cutoff)

    query = query.offset(offset).limit(limit)

    result = await db.execute(query)
    logs = result.scalars().all()

    # Get provider info for each log
    response = []
    providers_cache: dict[int, NotificationProvider | None] = {}

    for log in logs:
        if log.provider_id not in providers_cache:
            provider_result = await db.execute(
                select(NotificationProvider).where(NotificationProvider.id == log.provider_id)
            )
            providers_cache[log.provider_id] = provider_result.scalar_one_or_none()

        provider = providers_cache[log.provider_id]
        # A provider the caller can't see is not named, as it isn't listed, and
        # neither is its error, which names its kind and destination.
        hidden = not printer_scope.is_unrestricted and (
            provider is None or not _provider_in_scope(provider, printer_scope)
        )
        if hidden:
            provider = None
        response.append(
            NotificationLogResponse(
                id=log.id,
                provider_id=log.provider_id,
                provider_name=provider.name if provider else None,
                provider_type=provider.provider_type if provider else None,
                event_type=log.event_type,
                title=log.title,
                message=log.message,
                success=log.success,
                error_message=None if hidden else log.error_message,
                printer_id=log.printer_id,
                printer_name=log.printer_name,
                created_at=log.created_at,
            )
        )

    return response


@router.get("/logs/stats", response_model=NotificationLogStats)
async def get_notification_log_stats(
    days: int = Query(default=7, ge=1, le=90, description="Statistics for the last N days"),
    db: AsyncSession = Depends(get_db),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.NOTIFICATIONS_READ),
    printer_scope: PrinterScope = RequestPrinterScope,
):
    """Get notification log statistics."""
    cutoff = datetime.now(timezone.utc) - timedelta(days=days)
    # Counted over the entries the log list shows this caller.
    window = [NotificationLog.created_at >= cutoff]
    if (scope_clause := _log_scope_clause(printer_scope)) is not None:
        window.append(scope_clause)

    # Total counts
    total_result = await db.execute(select(func.count(NotificationLog.id)).where(*window))
    total = total_result.scalar() or 0

    success_result = await db.execute(
        select(func.count(NotificationLog.id)).where(*window, NotificationLog.success.is_(True))
    )
    success_count = success_result.scalar() or 0

    # By event type
    event_result = await db.execute(
        select(NotificationLog.event_type, func.count(NotificationLog.id))
        .where(*window)
        .group_by(NotificationLog.event_type)
    )
    by_event_type = {row[0]: row[1] for row in event_result.fetchall()}

    # By provider (need to join to get name)
    provider_result = await db.execute(
        select(NotificationProvider, func.count(NotificationLog.id))
        .join(NotificationProvider, NotificationLog.provider_id == NotificationProvider.id)
        .where(*window)
        .group_by(NotificationProvider.id)
    )
    # Providers the caller can't see are counted together, not by name.
    by_provider: dict[str, int] = {}
    for provider, count in provider_result.all():
        name = provider.name if _provider_in_scope(provider, printer_scope) else "Other"
        by_provider[name] = by_provider.get(name, 0) + count

    return NotificationLogStats(
        total=total,
        success_count=success_count,
        failure_count=total - success_count,
        by_event_type=by_event_type,
        by_provider=by_provider,
    )


@router.delete("/logs")
async def clear_notification_logs(
    older_than_days: int = Query(default=30, ge=1, description="Delete logs older than N days"),
    db: AsyncSession = Depends(get_db),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.NOTIFICATIONS_DELETE),
    printer_scope: PrinterScope = RequestPrinterScope,
):
    """Clear old notification logs."""
    cutoff = datetime.now(timezone.utc) - timedelta(days=older_than_days)

    statement = delete(NotificationLog).where(NotificationLog.created_at < cutoff)
    # A caller limited to some printers clears only those printers' entries.
    if (scope_clause := printer_scope.where_strict(NotificationLog.printer_id)) is not None:
        statement = statement.where(scope_clause)
    result = await db.execute(statement)
    await db.commit()

    deleted_count = result.rowcount
    logger.info("Deleted %s notification logs older than %s days", deleted_count, older_than_days)

    return {"deleted": deleted_count, "message": f"Deleted {deleted_count} logs older than {older_than_days} days"}


@router.get("/photos/{filename}")
async def get_notification_photo(filename: str):
    """Serve an ad-hoc notification snapshot to HA, Bark or Slack.

    They fetch this URL themselves with no session, so the unguessable
    filename is the credential and opens this one photo only -- see
    backend/app/utils/notification_photos.py. Anything that isn't a live
    photo of exactly that shape is a 404.
    """
    photo_path = find_notification_photo(filename)
    if photo_path is None:
        raise HTTPException(404, "Photo not found")

    return FileResponse(path=photo_path, media_type="image/jpeg", headers={"Cache-Control": "private, no-store"})


# ============================================================================
# Provider Instance Routes (parameterized - must come LAST)
# ============================================================================


# Messages from other applications -------------------------------------------

# Per caller, in memory: enough for any real app (Bambuddy Orders sends a few a
# day), and a buggy or hostile one can't flood the channels.
APP_MESSAGE_LIMIT = 20
APP_MESSAGE_WINDOW_SECONDS = 60
_app_message_times: dict[str, deque[float]] = defaultdict(deque)


def _app_sender(caller: ScopedCaller) -> tuple[str, str]:
    """(rate-limit key, name shown in the log) for whoever sends the message."""
    if caller.api_key is not None:
        return f"key:{caller.api_key.id}", caller.api_key.name
    if caller.user is not None:
        return f"user:{caller.user.id}", caller.user.username
    return "anonymous", "app"


def forget_app_senders(api_key_ids=(), user_ids=()) -> None:
    """Drop deleted keys' and users' message counts: SQLite gives an id to the
    next key or user created, which would start out throttled."""
    for key_id in api_key_ids:
        _app_message_times.pop(f"key:{key_id}", None)
    for user_id in user_ids:
        _app_message_times.pop(f"user:{user_id}", None)


def _check_app_message_rate(key: str) -> None:
    times = _app_message_times[key]
    cutoff = time.monotonic() - APP_MESSAGE_WINDOW_SECONDS
    while times and times[0] < cutoff:
        times.popleft()
    if len(times) >= APP_MESSAGE_LIMIT:
        raise HTTPException(status_code=429, detail="Too many messages; try again in a minute")
    times.append(time.monotonic())


@router.post("/app-message", response_model=AppMessageResult)
async def send_app_message(
    data: AppMessage,
    db: AsyncSession = Depends(get_db),
    caller: ScopedCaller = Depends(require_notification_send()),
    printer_scope: PrinterScope = RequestPrinterScope,
):
    """Send a message through every enabled channel that has "Messages from
    connected apps" on. For other applications, e.g. Bambuddy Orders; an API
    key needs the "Send notifications" permission. A caller limited to some
    printers reaches only the channels of those printers."""
    key, sender = _app_sender(caller)
    _check_app_message_rate(key)
    channels = await notification_service.on_app_message(
        db,
        sender=sender,
        title=data.title,
        message=data.message,
        url=data.url,
        allowed=lambda provider: _provider_in_scope(provider, printer_scope),
    )
    return AppMessageResult(channels=channels)


@router.get("/app-message/channels", response_model=list[AppMessageChannel])
async def app_message_channels(
    db: AsyncSession = Depends(get_db),
    _: ScopedCaller = Depends(require_notification_send()),
    printer_scope: PrinterScope = RequestPrinterScope,
):
    """The enabled channels that deliver app messages: names and types only,
    so an app can tell its user where its messages will arrive."""
    rows = await db.execute(
        select(NotificationProvider)
        .where(NotificationProvider.enabled.is_(True), NotificationProvider.on_app_message.is_(True))
        .order_by(NotificationProvider.name)
    )
    return [
        AppMessageChannel(name=p.name, provider_type=p.provider_type)
        for p in rows.scalars()
        if _provider_in_scope(p, printer_scope)
    ]


@router.get("/{provider_id}", response_model=NotificationProviderResponse)
async def get_notification_provider(
    provider_id: int,
    db: AsyncSession = Depends(get_db),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.NOTIFICATIONS_READ),
    printer_scope: PrinterScope = RequestPrinterScope,
):
    """Get a specific notification provider."""
    provider = await _load_provider(db, provider_id, printer_scope)

    return _provider_to_dict(provider)


@router.patch("/{provider_id}", response_model=NotificationProviderResponse)
async def update_notification_provider(
    provider_id: int,
    update_data: NotificationProviderUpdate,
    db: AsyncSession = Depends(get_db),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.NOTIFICATIONS_UPDATE),
    printer_scope: PrinterScope = RequestPrinterScope,
):
    """Update a notification provider."""
    provider = await _load_provider(db, provider_id, printer_scope)

    old_config = _stored_config(provider)
    old_type = provider.provider_type

    # Update only provided fields
    update_dict = update_data.model_dump(exclude_unset=True)
    if "printer_id" in update_dict:
        _ensure_target_in_scope(update_dict["printer_id"], printer_scope)
    await ensure_printers_exist(db, [update_dict.get("printer_id")])

    effective_type = update_dict.get("provider_type") or provider.provider_type
    effective_type = getattr(effective_type, "value", effective_type)
    if effective_type != old_type and update_dict.get("config") is None:
        raise HTTPException(status_code=400, detail="A new provider type needs its own configuration")
    if update_dict.get("config") is not None:
        try:
            update_dict["config"] = merge_config(old_type, old_config, effective_type, update_dict["config"])
        except SecretConfigError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from None
    if effective_type == "notify":
        try:
            effective_config = update_dict.get("config", old_config)
            if not isinstance(effective_config, dict):
                raise NotifyError("Notify! configuration must be an object")
            notify_credentials(effective_config)
        except NotifyError as exc:
            raise HTTPException(status_code=422, detail=str(exc)) from None

    for key, value in update_dict.items():
        if key == "config" and value is not None:
            setattr(provider, key, json.dumps(value))
        elif key == "provider_type" and value is not None:
            setattr(provider, key, value.value)
        else:
            setattr(provider, key, value)

    if old_type == "notify":
        effective_config = _stored_config(provider)
        remove_widgets = (
            provider.provider_type != "notify"
            or not provider.enabled
            or effective_config.get("lock_screen_widgets") is not True
        )
        if remove_widgets or provider.printer_id is not None:
            condition = NotificationLockScreenWidget.provider_id == provider_id
            if not remove_widgets:
                condition = condition & (NotificationLockScreenWidget.printer_id != provider.printer_id)
            # Persist opt-out before returning, even if the user immediately
            # opts back in. The worker finishes cleanup before creating anew.
            await db.execute(
                update(NotificationLockScreenWidget)
                .where(condition)
                .values(
                    state="deleting",
                    failures=0,
                    next_attempt_at=None,
                )
            )

    await db.commit()
    await db.refresh(provider)

    response = _provider_to_dict(provider)
    new_config = _stored_config(provider)
    changed = old_type != provider.provider_type or any(
        old_config.get(key) != new_config.get(key) for key in ("device_id", "token")
    )
    # Refresh opened a new read transaction; close it before remote cleanup.
    await db.commit()
    if old_type == "notify" and changed:
        await notify_live_activities.schedule_cleanup(provider_id, old_config)
        await notify_widgets.schedule_cleanup(provider_id, old_config)
    if old_type == "notify" or provider.provider_type == "notify":
        _notify_providers_changed()
    logger.info("Updated notification provider: %s", provider.name)
    await _resync_reaction_poller()

    return response


@router.delete("/{provider_id}")
async def delete_notification_provider(
    provider_id: int,
    db: AsyncSession = Depends(get_db),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.NOTIFICATIONS_DELETE),
    printer_scope: PrinterScope = RequestPrinterScope,
):
    """Delete a notification provider."""
    provider = await _load_provider(db, provider_id, printer_scope)

    name = provider.name
    notify = await delete_provider(db, provider)
    await db.commit()
    await after_providers_deleted(notify=notify)

    logger.info("Deleted notification provider: %s", name)

    return {"message": f"Notification provider '{name}' deleted"}


@router.post("/{provider_id}/test", response_model=NotificationTestResponse)
async def test_notification_provider(
    provider_id: int,
    db: AsyncSession = Depends(get_db),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.NOTIFICATIONS_UPDATE),
    printer_scope: PrinterScope = RequestPrinterScope,
):
    """Send a test notification using an existing provider."""
    provider = await _load_provider(db, provider_id, printer_scope)

    config = json.loads(provider.config) if isinstance(provider.config, str) else provider.config
    success, message = await notification_service.send_test_notification(
        provider.provider_type, config, db, attach_photo=provider.attach_photo
    )

    # Update provider status
    if success:
        provider.last_success = datetime.now(timezone.utc)
    else:
        provider.last_error = message
        provider.last_error_at = datetime.now(timezone.utc)

    await db.commit()

    return NotificationTestResponse(success=success, message=message)