siftlode/backend/app/email.py

"""Outbound email for onboarding (access requests + approval notices).

Gmail SMTP + App Password by default. Deliberately fail-soft: if SMTP isn't configured
or a send errors, we log and return False so the caller can fall back to in-app
notifications — email is never load-bearing for the app's core flow.

Fail-soft must not mean invisible, though: every drop also leaves an in-app notice in the admins'
inbox (see `_alert_admins`). An instance whose SMTP was never filled in silently swallows
verification links, password resets and access-request notices, and the operator's first clue used
to be a person telling them nothing arrived.
"""

import html
import logging
import re
import smtplib
import ssl
from email.message import EmailMessage
from email.utils import formatdate

from sqlalchemy import select
from sqlalchemy.orm import Session

from app import sysconfig
from app.config import settings
from app.db import SessionLocal
from app.models import User
from app.notifications import create_notification
from app.ratelimit import RateLimiter

log = logging.getLogger("siftlode.email")

# One admin notice an hour per reason. A relay that is down fails every send, and the inbox must
# stay readable; the log keeps the full record either way.
_alert_throttle = RateLimiter(max_events=1, window_seconds=3600)


def _alert_admins(reason: str, subject: str, recipients: list[str]) -> None:
    """Record an in-app notice that a message did NOT go out. Best effort in every direction: it
    opens its own session (callers are often background tasks), swallows its own errors, and never
    reports failure to the caller — this is the fallback path, it must not become a new way to
    break a send."""
    if not _alert_throttle.allow(reason):
        return
    try:
        with SessionLocal() as db:
            admins = (
                db.execute(
                    select(User.id).where(
                        User.role == "admin",
                        User.is_active.is_(True),
                        User.is_suspended.is_(False),
                    )
                )
                .scalars()
                .all()
            )
            if not admins:
                # Nothing was recorded, so don't let this cost the next hour of alerts on the very
                # path whose job is to stop drops going unseen. Same on the failure branch below.
                _alert_throttle.reset()
                return
            for uid in admins:
                create_notification(
                    db,
                    user_id=uid,
                    type="email_undelivered",
                    # English fallback; the inbox renders a translated version from `data`.
                    title="Email could not be sent",
                    body=subject,
                    data={"reason": reason, "subject": subject, "recipients": recipients},
                    commit=False,
                )
            db.commit()
    except Exception:
        _alert_throttle.reset()
        log.exception("Could not record the undelivered-email notice (%r)", subject)


def _read_smtp(db: Session) -> dict:
    return {
        "host": sysconfig.get_str(db, "smtp_host"),
        "port": sysconfig.get_int(db, "smtp_port"),
        "user": sysconfig.get_str(db, "smtp_user"),
        "password": sysconfig.get_str(db, "smtp_password"),
        "from": sysconfig.get_str(db, "smtp_from"),
    }


def _smtp(db: Session | None = None) -> dict:
    """Effective SMTP config (admin DB override over env defaults). Opens its own short session when
    the caller has none, because email is usually sent from a BackgroundTask outside a request's db.
    A caller that IS inside a request should pass its session: reading the password decrypts a Fernet
    value, and /api/me polls this for every admin tab."""
    if db is not None:
        return _read_smtp(db)
    with SessionLocal() as own:
        return _read_smtp(own)


def email_enabled(db: Session | None = None) -> bool:
    c = _smtp(db)
    return bool(c["host"] and c["user"] and c["password"])


def _admin_contact() -> str | None:
    return next(iter(sorted(settings.admin_email_set)), None)


# A bare URL in a text/plain mail is only clickable if the CLIENT decides to linkify it. Gmail,
# Outlook and Apple Mail do; plenty of lighter web clients render the body verbatim and leave the
# link as dead text — which, for a verification or password-reset mail, is the whole message. So
# every mail also carries an HTML alternative built from the SAME plain body (below), rather than
# per-template HTML that would have to be kept in sync with it.
#
# `[^\s<]` stops at whitespace and at the `<` of any following markup; the trailing-punctuation trim
# keeps a sentence-final period out of the href. `;` and `:` are deliberately NOT trimmed — escaping
# turns `&` into `&amp;`, so a query string can legitimately end in a semicolon.
_URL_RE = re.compile(r"https?://[^\s<]+")


def _html_body(text: str) -> str:
    """Render a plain-text mail body as minimal HTML: escaped, blank-line-separated paragraphs, with
    real links. Escaping happens FIRST and the linkifier then runs over the escaped text — bodies
    interpolate untrusted values (a requester's email address, the operator contact), so no caller
    can inject markup, and `&` inside a URL is already `&amp;` by the time it reaches the href, which
    is what an HTML attribute wants anyway."""

    def _anchor(m: re.Match[str]) -> str:
        url = m.group(0)
        trail = ""
        while url and url[-1] in ".,!?)":
            url, trail = url[:-1], url[-1] + trail
        if not url:
            return trail
        return f'<a href="{url}">{url}</a>{trail}'

    paragraphs = [p for p in html.escape(text).split("\n\n") if p.strip()]
    rendered = "".join(
        f"<p>{_URL_RE.sub(_anchor, p).replace(chr(10), '<br>')}</p>" for p in paragraphs
    )
    # Inline styles only — many clients drop <style> blocks. No images, no external CSS, no tracking:
    # the HTML part exists to make the link clickable, not to become a template to maintain.
    return (
        '<html><body style="margin:0;padding:16px;font-family:system-ui,-apple-system,'
        "'Segoe UI',sans-serif;font-size:15px;line-height:1.6\">"
        f'<div style="max-width:560px">{rendered}</div>'
        "</body></html>"
    )


def _send(to: list[str], subject: str, body: str, reply_to: str | None = None) -> bool:
    recipients = [e for e in to if e]
    if not recipients:
        return False
    c = _smtp()
    if not (c["host"] and c["user"] and c["password"]):
        log.info("SMTP not configured; skipping email %r to %s", subject, recipients)
        _alert_admins("not_configured", subject, recipients)
        return False
    msg = EmailMessage()
    msg["Subject"] = subject
    msg["From"] = c["from"] or c["user"]
    msg["To"] = ", ".join(recipients)
    # A real Date and a Reply-To (so it's a conversation, not a no-reply blast) both
    # nudge spam filters the right way; reputation still does most of the work.
    msg["Date"] = formatdate(localtime=True)
    if reply_to:
        msg["Reply-To"] = reply_to
    # text/plain stays the primary part (it is the source of truth, and a text-only client still
    # gets the full message); the HTML alternative below is what makes the link clickable everywhere.
    msg.set_content(body)
    msg.add_alternative(_html_body(body), subtype="html")
    try:
        with smtplib.SMTP(c["host"], c["port"], timeout=20) as s:
            s.starttls(context=ssl.create_default_context())
            s.login(c["user"], c["password"])
            s.send_message(msg)
        log.info("Sent email %r to %s", subject, recipients)
        return True
    except Exception:
        log.exception("SMTP send failed (%r to %s)", subject, recipients)
        _alert_admins("send_failed", subject, recipients)
        return False


def send_access_approved(email: str) -> bool:
    admin = _admin_contact()
    body = (
        "Hi,\n\n"
        "Your request to use Siftlode has been approved — welcome aboard.\n\n"
        "Open Siftlode and sign in, and your YouTube subscriptions feed will be ready:\n\n"
        f"{settings.app_base}\n\n"
        "If you weren't expecting this, you can ignore the message.\n\n"
        "— Siftlode" + (f"\n\nQuestions? Just reply to this email ({admin})." if admin else "")
    )
    return _send([email], "You're in — your Siftlode access is approved", body, reply_to=admin)


def send_access_denied(email: str) -> bool:
    admin = _admin_contact()
    body = (
        "Hi,\n\n"
        "Thanks for your interest in Siftlode. After review, your access request wasn't approved "
        "this time.\n\n"
        "If you think this was a mistake, you're welcome to reply and let us know.\n\n"
        "— Siftlode" + (f"\n\nQuestions? Just reply to this email ({admin})." if admin else "")
    )
    return _send([email], "About your Siftlode access request", body, reply_to=admin)


def send_role_changed(to: str, role: str, operator: str | None) -> bool:
    line = (
        "You've been granted admin access on Siftlode — you can now manage settings, the "
        "scheduler and users."
        if role == "admin"
        else "Your admin access on Siftlode has been removed — you're now a regular user."
    )
    contact = f"\n\nQuestions? Contact the operator at {operator}." if operator else ""
    body = "Hi,\n\n" + line + contact + "\n\n— Siftlode"
    return _send([to], "Your Siftlode role has changed", body, reply_to=operator)


def send_account_unsuspended(to: str, operator: str | None) -> bool:
    contact = f"\n\nQuestions? Contact the operator at {operator}." if operator else ""
    body = (
        "Hi,\n\n"
        "Your Siftlode account has been reinstated — the suspension was lifted and you can sign "
        "in again." + contact + "\n\n— Siftlode"
    )
    return _send([to], "Your Siftlode account has been reinstated", body, reply_to=operator)


def send_account_deleted(to: str, operator: str | None) -> bool:
    contact = (
        f"\n\nIf you didn't request this, contact the Siftlode operator at {operator}."
        if operator
        else ""
    )
    body = (
        "Hi,\n\n"
        "Your Siftlode account and all associated data — subscriptions, tags, watch history, "
        "playlists and settings — have been permanently deleted. Nothing is retained.\n\n"
        "Thanks for having used Siftlode." + contact + "\n\n— Siftlode"
    )
    return _send([to], "Your Siftlode account has been deleted", body, reply_to=operator)


def send_account_suspended(to: str, operator: str | None) -> bool:
    contact = (
        f"If you think this is a mistake or have questions, contact the Siftlode operator "
        f"at {operator}.\n\n"
        if operator
        else "If you think this is a mistake, please contact the Siftlode operator.\n\n"
    )
    body = (
        "Hi,\n\n"
        "A sign-in to your Siftlode account was just attempted, but the account is currently "
        "suspended, so access was denied.\n\n" + contact + "— Siftlode"
    )
    return _send([to], "Your Siftlode account is suspended", body, reply_to=operator)


def send_admin_new_request(admins: list[str], requester: str) -> bool:
    # The link deep-links straight to the admin Access-requests page (the SPA reads ?admin=…). The
    # requester address is spelled out (mail clients auto-link both the URL and the address), and the
    # reply note is accurate: _send sets Reply-To to the requester, so a reply reaches THEM — not the
    # sending mailbox. (The old copy said "Settings → Account", which was the wrong menu.)
    approve_url = f"{settings.app_base}/?admin=access-requests"
    body = (
        f"{requester} just requested access to Siftlode.\n\n"
        f"Approve or deny it here:\n{approve_url}\n"
        f"(or in Siftlode: open Users → Access requests)\n\n"
        f"Requester's email: {requester}\n"
        f"Replying to this message goes straight to the requester (it's the Reply-To address).\n"
    )
    return _send(admins, f"Siftlode access request from {requester}", body, reply_to=requester)


def send_verify_email(to: str, link: str) -> bool:
    body = (
        "Hi,\n\n"
        "Confirm your email to finish creating your Siftlode account:\n\n"
        f"{link}\n\n"
        "If you didn't sign up, you can ignore this message.\n\n"
        "— Siftlode"
    )
    return _send([to], "Confirm your Siftlode email", body)


def send_password_reset(to: str, link: str) -> bool:
    body = (
        "Hi,\n\n"
        "Use this link to set a new Siftlode password (it expires in an hour):\n\n"
        f"{link}\n\n"
        "If you didn't request this, you can ignore this message — your password is unchanged.\n\n"
        "— Siftlode"
    )
    return _send([to], "Reset your Siftlode password", body)


def send_test(to: str) -> bool:
    body = (
        "This is a test email from Siftlode.\n\n"
        "If you're reading this, your SMTP settings work.\n\n"
        "— Siftlode"
    )
    return _send([to], "Siftlode SMTP test", body)