ServerKit/backend/app/notifications/digest.py

"""Notification digest rollups — per-user cadence.

Instead of one email per event, a user can opt into a *digest*: the bus routes
digestable deliveries (and quiet-hours catch-ups) onto the
``notification_deliveries`` table with status ``queued_digest`` (never enqueued),
and this job periodically groups a user's held deliveries into ONE branded email,
sends it, and flips those rows to ``sent`` with the digest's message id.

Cadence lives on ``NotificationPreferences.digest_cadence`` ('off'|'daily'|
'weekly'); ``digest_last_sent_at`` records the last send so the hourly tick only
mails a user once their window has elapsed.

Guards (Decision 3 / #9): an empty digest never sends (and never stamps the
clock, so nothing is silently swallowed); a one-item digest borrows that item's
own subject line rather than shaming the reader with a "1 update" email.

Entry points:
  - ``run_due(now=None)``      — the hourly tick; mails every user whose window
                                 elapsed. This is what the ScheduledJob calls.
  - ``send_user_digest(uid)``  — build + send one user's digest now.
  - ``run_digest_job(work_unit=None)`` — job-handler-shaped wrapper for run_due.
"""
import logging
from collections import OrderedDict
from datetime import datetime, timedelta

from app import db
from app.models.notification_preferences import NotificationPreferences
from app.models.user import User
from app.notifications import rendering
from app.notifications.models import Notification, NotificationDelivery

logger = logging.getLogger(__name__)

# How long between digests for each cadence.
_CADENCE_DELTA = {
    'daily': timedelta(days=1),
    'weekly': timedelta(days=7),
}

# Preference categories -> human labels for the digest section headers.
_CATEGORY_LABELS = {
    'system': 'System',
    'security': 'Security',
    'backups': 'Backups',
    'apps': 'Apps',
}
# Deterministic section order (unknown categories append, alpha).
_CATEGORY_ORDER = ['security', 'backups', 'apps', 'system']


# --------------------------------------------------------------------------- #
# Selection
# --------------------------------------------------------------------------- #

def pending_for_user(user_id):
    """This user's held (``queued_digest``) email deliveries, oldest first."""
    return (NotificationDelivery.query
            .filter_by(recipient_user_id=user_id,
                       channel=NotificationDelivery.CHANNEL_EMAIL,
                       status=NotificationDelivery.STATUS_QUEUED_DIGEST)
            .order_by(NotificationDelivery.created_at.asc())
            .all())


def _cadence_delta(cadence):
    return _CADENCE_DELTA.get((cadence or 'off').strip().lower())


def is_due(prefs, now=None):
    """Whether ``prefs``' digest window has elapsed (or never sent)."""
    delta = _cadence_delta(prefs.digest_cadence)
    if delta is None:  # 'off' or unknown -> never due
        return False
    if prefs.digest_last_sent_at is None:
        return True
    return (now or datetime.utcnow()) - prefs.digest_last_sent_at >= delta


# --------------------------------------------------------------------------- #
# Build + render
# --------------------------------------------------------------------------- #

def build_digest(user, deliveries):
    """Group ``deliveries`` (with their notifications) into digest sections.

    Returns ``{'subject', 'groups', 'total', 'user'}`` or None if nothing to
    say (all deliveries orphaned of their notification)."""
    buckets = OrderedDict()
    total = 0
    first_title = None
    for delivery in deliveries:
        notification = delivery.notification
        if notification is None:
            continue
        total += 1
        if first_title is None:
            first_title = notification.title
        category = notification.category or 'system'
        buckets.setdefault(category, []).append({
            'severity': notification.severity or 'info',
            'title': notification.title,
            'when': notification.created_at.strftime('%b %d, %H:%M') if notification.created_at else None,
            'action_url': notification.action_path or None,
        })

    if total == 0:
        return None

    ordered = sorted(
        buckets.items(),
        key=lambda kv: (_CATEGORY_ORDER.index(kv[0]) if kv[0] in _CATEGORY_ORDER
                        else len(_CATEGORY_ORDER), kv[0]),
    )
    groups = [{'label': _CATEGORY_LABELS.get(cat, cat.replace('_', ' ').title()),
               'category': cat, 'items': items}
              for cat, items in ordered]

    # Single-item digests borrow the item's own subject (#9 — no "1 update" mail).
    if total == 1:
        subject = first_title
    else:
        subject = f'ServerKit digest — {total} updates'

    return {'subject': subject, 'groups': groups, 'total': total, 'user': user}


def render_digest(digest, manage_url=None):
    """Render a built digest to ``{'subject', 'html', 'text'}``."""
    user = digest['user']
    recipient = {'name': (user.username or user.email) if user else None,
                 'email': user.email if user else None}
    ctx = rendering._build_context(
        subject=digest['subject'], severity='info', data={},
        recipient=recipient, urls={'manage': manage_url}, hostname=None,
    )
    ctx['groups'] = digest['groups']
    ctx['total'] = digest['total']
    try:
        html = rendering.env().get_template('email/digest.html').render(**ctx)
    except Exception as exc:  # never let a template bug swallow the digest
        logger.warning('Digest template failed (%s); using generic body', exc)
        html = rendering.env().get_template('email/generic.html').render(**ctx)
    return {'subject': digest['subject'], 'html': html, 'text': _text_twin(digest)}


def _text_twin(digest):
    """Programmatic plain-text alternative (like rendering's fallback)."""
    lines = [digest['subject'], '=' * min(len(digest['subject']), 60), '']
    for group in digest['groups']:
        lines.append(group['label'].upper())
        for item in group['items']:
            when = f" ({item['when']})" if item.get('when') else ''
            lines.append(f"  - [{item['severity']}] {item['title']}{when}")
        lines.append('')
    lines.append('-' * 40)
    lines.append('Manage your digest cadence in ServerKit → Settings → Notifications.')
    return '\n'.join(lines)


# --------------------------------------------------------------------------- #
# Transport (seam — tests override this)
# --------------------------------------------------------------------------- #

def _transmit(to_addr, rendered):
    """Send the digest email. Returns ``(success, message_id, error)``.

    Prefers a configured EmailProviderConnection; otherwise falls back to the
    legacy notifications.json SMTP path (reusing the email channel adapter)."""
    from app.notifications.providers import EmailProviderService
    try:
        provider = EmailProviderService.default_provider()
    except Exception:
        provider = None
    if provider is not None:
        result = EmailProviderService.send(
            provider, to_addr, rendered['subject'], rendered['html'], rendered['text'])
        return bool(result.get('success')), result.get('message_id'), result.get('error')

    from app.notifications.channels.email import EmailAdapter
    res = EmailAdapter()._send_legacy_smtp(to_addr, rendered)
    return res.ok, res.message_id, res.error


# --------------------------------------------------------------------------- #
# Send
# --------------------------------------------------------------------------- #

def _digest_address(user, prefs):
    return (prefs.email if prefs else None) or (user.email if user else None)


def send_user_digest(user_id, now=None, stamp_clock=True):
    """Build + send one user's digest. Returns a result dict.

    On success: the digested deliveries flip to ``sent`` (carrying the digest's
    message id) and ``digest_last_sent_at`` is stamped. An empty digest is a
    no-op that neither sends nor stamps the clock."""
    now = now or datetime.utcnow()
    user = User.query.get(user_id)
    if user is None:
        return {'sent': False, 'reason': 'no user', 'count': 0}

    deliveries = pending_for_user(user_id)
    if not deliveries:
        return {'sent': False, 'reason': 'empty', 'count': 0}

    prefs = NotificationPreferences.get_or_create(user_id)
    to_addr = _digest_address(user, prefs)
    if not to_addr:
        return {'sent': False, 'reason': 'no email address', 'count': len(deliveries)}

    digest = build_digest(user, deliveries)
    if digest is None:  # deliveries all orphaned of notifications
        return {'sent': False, 'reason': 'empty', 'count': 0}

    rendered = render_digest(digest)
    success, message_id, error = _transmit(to_addr, rendered)
    if not success:
        return {'sent': False, 'reason': error or 'send failed',
                'count': len(deliveries)}

    sent_at = datetime.utcnow()
    for delivery in deliveries:
        delivery.status = NotificationDelivery.STATUS_SENT
        delivery.sent_at = sent_at
        delivery.provider_message_id = message_id
        delivery.error = None
    if stamp_clock:
        prefs.digest_last_sent_at = now
    db.session.commit()
    return {'sent': True, 'count': len(deliveries), 'subject': digest['subject'],
            'message_id': message_id}


def run_due(now=None):
    """Hourly tick: send a digest to every user whose cadence window elapsed."""
    now = now or datetime.utcnow()
    rows = (NotificationPreferences.query
            .filter(NotificationPreferences.digest_cadence.in_(tuple(_CADENCE_DELTA)))
            .all())
    sent = skipped = 0
    for prefs in rows:
        if not is_due(prefs, now=now):
            continue
        result = send_user_digest(prefs.user_id, now=now)
        if result.get('sent'):
            sent += 1
        else:
            skipped += 1
    return {'sent': sent, 'skipped': skipped, 'candidates': len(rows)}


def run_digest_job(work_unit=None):
    """ScheduledJob handler shape (kind ``notifications.digest.run``)."""
    return run_due()