Files
rfc-app/backend/app/email.py
T
Ben Stull 281a844513 v0.18.0 Slice 4: outbound_emails audit table + admin endpoint
Per the v0.18.0 email + webhook hygiene proposal §3:

  * `backend/migrations/020_outbound_emails.sql` — new table
    capturing every send attempt: to_address, from_address,
    subject, kind, sent_at, status ('sent' | 'failed' | 'deferred'
    | 'bounced'), error, notification_id (FK), message_id (for
    Slice 5 bounce correlation).
  * `email.record_outbound()` — best-effort write helper every
    send path calls. Status='sent' on SMTP success, 'failed' on
    exception (with class + message in `error`), 'deferred' on
    the dev-fallback path where SMTP_HOST is unset (the send
    didn't happen but the row records the attempt).
  * `email._deliver` (watcher notifications + bundles), `digest.py`,
    `email_otc.py`, `email_invite.py` — every send path now records.
  * `GET /api/admin/outbound-emails` — admin-only listing,
    filterable by kind / status / to_address. Answers "did this
    person ever get their invite?" without grepping VM logs. No
    admin UI in v0.18.0; operator queries via curl + jq for now.

7 new integration tests covering: OTC / invite / notification all
write rows with matching Message-ID; admin endpoint lists,
filters by kind, filters by to_address (case-insensitive),
refuses non-admins.

Full suite: 291 passed.
2026-05-28 07:32:31 -07:00

614 lines
22 KiB
Python

"""§15.4: the email loop.
The transactional-email adapter is a thin wrapper over SMTP. When SMTP
credentials aren't configured the adapter falls back to logging the
envelope to stdout — sufficient for dev and for the integration tests,
which assert on the log buffer rather than spinning up a mail server.
Per §15.4: opt-in per category (with category-specific defaults), one-
click unsubscribe per category via signed URL, single non-spoofing From
identity, body mirrors the inbox row verbatim. Bounces and complaints
route to a global opt-out per the same section.
Quiet hours per §15.8 hold the send (the notification row still lands;
the email defers) until the window ends. The release-from-hold pass is
the same `flush_pending` the digest job calls; it bundles into a single
"Activity while you were away" email when more than a threshold
accumulated, otherwise sending individually.
"""
from __future__ import annotations
import json
import logging
import os
import smtplib
from dataclasses import dataclass
from datetime import datetime, time, timezone
from email.utils import formataddr
from itertools import groupby
from typing import Any
from urllib.parse import urlencode
from itsdangerous import BadSignature, URLSafeSerializer
from . import db
from .email_envelope import build_envelope
log = logging.getLogger(__name__)
# Buffer of outbound envelopes for test inspection. The integration tests
# read from this rather than spinning up a real SMTP server. Production
# leaves it as an unbounded list (it's never read), which is fine at
# v1 volumes; if it ever becomes a memory issue, we cap it then.
_SENT: list[dict] = []
def sent_envelopes() -> list[dict]:
return list(_SENT)
def reset_sent_envelopes() -> None:
_SENT.clear()
# ---------------------------------------------------------------------------
# Configuration — read from env at call time so tests can monkeypatch
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class EmailConfig:
smtp_host: str
smtp_port: int
smtp_user: str
smtp_password: str
smtp_starttls: bool
from_address: str
from_name: str
app_url: str
bundle_threshold: int
enabled: bool
unsubscribe_mailto: str
@classmethod
def from_env(cls) -> "EmailConfig":
host = os.environ.get("SMTP_HOST", "").strip()
return cls(
smtp_host=host,
smtp_port=int(os.environ.get("SMTP_PORT", "587")),
smtp_user=os.environ.get("SMTP_USER", "").strip(),
smtp_password=os.environ.get("SMTP_PASSWORD", ""),
smtp_starttls=os.environ.get("SMTP_STARTTLS", "1") not in ("0", "false", "False", ""),
from_address=os.environ.get("EMAIL_FROM", "notifications@wiggleverse.local").strip(),
from_name=os.environ.get("EMAIL_FROM_NAME", "Wiggleverse").strip(),
app_url=os.environ.get("APP_URL", "http://localhost:8000").rstrip("/"),
bundle_threshold=int(os.environ.get("EMAIL_BUNDLE_THRESHOLD", "5")),
enabled=os.environ.get("EMAIL_ENABLED", "1") not in ("0", "false", "False"),
# v0.18.0: the `List-Unsubscribe: <mailto:…>` target on
# invite + notification mail. Defaults to the From
# address when unset; a deployment can route opt-out
# mail to a separate mailbox (e.g., a humans-monitored
# account distinct from the no-reply notifications
# sender) by setting this explicitly.
unsubscribe_mailto=os.environ.get(
"EMAIL_UNSUBSCRIBE_MAILTO",
os.environ.get("EMAIL_FROM", "notifications@wiggleverse.local"),
).strip(),
)
# Signed-URL token for §15.4 one-click unsubscribe. The signer is
# scoped to (user_id, category) so the URL is idempotent and revocable
# by rotating SECRET_KEY.
def _signer() -> URLSafeSerializer:
secret = os.environ.get("SECRET_KEY", "")
if not secret:
raise RuntimeError("SECRET_KEY required for email signing")
return URLSafeSerializer(secret, salt="email-unsubscribe")
def make_unsubscribe_url(user_id: int, category: str) -> str:
"""Build the §15.4 per-category one-click URL.
`category` is one of `personal-direct`, `structural`,
`admin-actionable` (the three per-category flags) or `all`
(v0.18.0: the bundle path, which sets `email_opt_out_all = 1`
because a bundle covers multiple categories and a per-category
opt-out wouldn't honor the user's intent).
"""
cfg = EmailConfig.from_env()
token = _signer().dumps({"u": user_id, "c": category})
qs = urlencode({"t": token})
return f"{cfg.app_url}/api/email/unsubscribe?{qs}"
def verify_unsubscribe_token(token: str) -> tuple[int, str]:
"""Returns (user_id, category) or raises BadSignature."""
data = _signer().loads(token)
return int(data["u"]), str(data["c"])
# ---------------------------------------------------------------------------
# Category routing
# ---------------------------------------------------------------------------
# Map §15.1 event_kinds to one of the four §15.4 categories. The 'churn'
# category never emails per §15.4 — naming the refusal in settings is
# more honest than silently omitting the toggle.
_EVENT_TO_CATEGORY: dict[str, str] = {
"proposal_merged": "personal-direct",
"proposal_declined": "personal-direct",
"proposal_opened_on_watched_topic": "structural",
"pr_opened": "structural",
"pr_merged": "structural",
"pr_withdrawn": "structural",
"pr_commit_added": "churn",
"pr_review_thread_new": "structural",
"pr_review_thread_reply": "personal-direct",
"pr_conflict_with_main": "structural",
"chat_message_in_participated_thread": "churn",
"chat_reply_to_my_message": "personal-direct",
"change_proposed_on_edited_passage": "personal-direct",
"flag_dropped_on_watched_rfc": "structural",
"flag_resolved_on_my_flag": "personal-direct",
"contribute_grant_added": "personal-direct",
"contribute_grant_revoked": "personal-direct",
"graduation_complete": "personal-direct",
"super_draft_graduation_ready": "admin-actionable",
"claim_opened": "structural",
# v0.9.0: roadmap item #7. A fresh beta-access request lands as
# an admin-actionable signal so it consults `email_admin_actionable`
# and reaches owners/admins only.
"new_beta_request": "admin-actionable",
}
def category_for(event_kind: str, fallback_category: str) -> str:
return _EVENT_TO_CATEGORY.get(event_kind, fallback_category)
def _user_wants_email(user_row: Any, category: str) -> bool:
"""Apply the §15.4 per-category toggle. Churn always returns False
(the toggle is permanently off per the spec). Global opt-out
(`email_opt_out_all`) suppresses every category."""
if user_row["email_opt_out_all"]:
return False
if category == "churn":
return False
if category == "personal-direct":
return bool(user_row["email_personal_direct"])
if category == "structural":
return bool(user_row["email_watched_structural"])
if category == "admin-actionable":
if user_row["role"] not in ("owner", "admin"):
return False
return bool(user_row["email_admin_actionable"])
return False
# ---------------------------------------------------------------------------
# Quiet hours (§15.8)
# ---------------------------------------------------------------------------
def _in_quiet_hours(user_row: Any) -> bool:
start = user_row["notification_quiet_hours_start"]
end = user_row["notification_quiet_hours_end"]
tz_name = user_row["notification_quiet_hours_timezone"]
if not (start and end and tz_name):
return False
try:
from zoneinfo import ZoneInfo
tz = ZoneInfo(tz_name)
except Exception:
return False
now_local = datetime.now(tz).time()
start_t = _parse_hhmm(start)
end_t = _parse_hhmm(end)
if start_t is None or end_t is None:
return False
if start_t <= end_t:
return start_t <= now_local < end_t
# Wraps midnight (e.g. 22:00 → 07:00).
return now_local >= start_t or now_local < end_t
def _parse_hhmm(text: str) -> time | None:
try:
hh, mm = text.split(":", 1)
return time(int(hh), int(mm))
except (ValueError, AttributeError):
return None
# ---------------------------------------------------------------------------
# The dispatch entry point — called from notify._schedule_email
# ---------------------------------------------------------------------------
def maybe_send(
*,
notif_id: int,
recipient_user_id: int,
event_kind: str,
category: str,
payload: dict,
) -> None:
"""Apply §15.4's per-category opt-in and §15.8's quiet-hours hold,
then send. The notification row still landed in `notifications`
regardless; this only governs the out-of-band reach."""
user = db.conn().execute(
"""
SELECT id, email, display_name, role,
email_personal_direct, email_watched_structural, email_admin_actionable,
email_opt_out_all,
notification_quiet_hours_start, notification_quiet_hours_end,
notification_quiet_hours_timezone
FROM users WHERE id = ?
""",
(recipient_user_id,),
).fetchone()
if user is None or not user["email"]:
return
effective_category = category_for(event_kind, category)
if not _user_wants_email(user, effective_category):
return
if _in_quiet_hours(user):
# §15.8: hold during the window. The row's email_sent_at stays
# null; the digest's exclusion rule 1 won't kick in yet. The
# `flush_pending` pass at window end picks it up.
return
_send_one(user, notif_id, payload, effective_category)
def _send_one(user: Any, notif_id: int, payload: dict, category: str) -> None:
cfg = EmailConfig.from_env()
if not cfg.enabled:
return
subject = _subject(payload)
body = _body(payload, user["id"], category, cfg)
# v0.18.0: notification mail is bulk-adjacent (a watcher can
# accumulate dozens of structural events on a busy RFC), so it
# carries the full one-click unsubscribe — Gmail and Yahoo
# require this for senders at OHM's volume tier per RFC 8058.
unsubscribe_url = make_unsubscribe_url(user["id"], category)
sent = _deliver(
cfg,
user["email"],
subject,
body,
unsubscribe_mailto=cfg.unsubscribe_mailto,
unsubscribe_url=unsubscribe_url,
kind="notification",
notification_id=notif_id,
)
if not sent:
return
db.conn().execute(
"UPDATE notifications SET email_sent_at = datetime('now') WHERE id = ?",
(notif_id,),
)
def _subject(payload: dict) -> str:
rfc_title = payload.get("rfc_title") or payload.get("rfc_slug") or ""
summary = payload.get("summary") or payload.get("event_kind") or ""
if rfc_title:
return f"[Wiggleverse] {summary}{rfc_title}".strip()
return f"[Wiggleverse] {summary}".strip()
def _body(payload: dict, user_id: int, category: str, cfg: EmailConfig) -> str:
summary = payload.get("summary") or ""
actor = payload.get("actor_display") or "the app"
rfc_title = payload.get("rfc_title") or payload.get("rfc_slug") or "this RFC"
when = payload.get("created_at") or ""
link = _deep_link(payload, cfg)
unsub = make_unsubscribe_url(user_id, category)
manage = f"{cfg.app_url}/settings/notifications"
return (
f"{summary}\n\n"
f"by {actor} on {rfc_title} · {when}\n\n"
f"{link}\n\n"
f"---\n"
f"Unsubscribe from this category: {unsub}\n"
f"Manage all preferences: {manage}\n"
)
def _deep_link(payload: dict, cfg: EmailConfig) -> str:
slug = payload.get("rfc_slug")
pr = payload.get("pr_number")
branch = payload.get("branch_name")
event_kind = payload.get("event_kind")
# v0.9.0: framework-scoped admin signals link to the admin
# surface, not /rfc/... The `new_beta_request` event is the
# canonical example; future framework-scoped admin events
# may reuse the same branch.
if event_kind == "new_beta_request":
return f"{cfg.app_url}/admin/users"
if slug and pr:
return f"{cfg.app_url}/rfc/{slug}/pr/{pr}"
if slug and branch:
return f"{cfg.app_url}/rfc/{slug}?branch={branch}"
if slug:
return f"{cfg.app_url}/rfc/{slug}"
return cfg.app_url
def _deliver(
cfg: EmailConfig,
to_address: str,
subject: str,
body: str,
*,
unsubscribe_mailto: str | None = None,
unsubscribe_url: str | None = None,
kind: str = "notification",
notification_id: int | None = None,
) -> bool:
"""Build the envelope via the shared `build_envelope` helper and
hand it to SMTP.
The `_SENT` buffer carries the helper's `EmailMessage` under
`message` plus the legacy `to`/`from`/`subject`/`body` keys for
backward-compatibility with tests that read those directly.
Newer tests can assert on the header surface by inspecting
`envelope["message"]`.
v0.18.0 Slice 4: also writes one row to `outbound_emails`
capturing the attempt. status='sent' on success, 'failed' on
SMTP exception, 'deferred' on the dev-fallback path (no
SMTP_HOST configured — the send didn't happen, but the row
records the attempt so the admin endpoint can answer "did the
framework try?").
"""
msg = build_envelope(
to_address=to_address,
from_address=cfg.from_address,
from_name=cfg.from_name,
subject=subject,
body_plain=body,
unsubscribe_mailto=unsubscribe_mailto,
unsubscribe_url=unsubscribe_url,
)
envelope = {
"to": to_address,
"from": formataddr((cfg.from_name, cfg.from_address)),
"subject": subject,
"body": body,
"message": msg,
"kind": kind,
}
_SENT.append(envelope)
message_id = msg["Message-ID"]
if not cfg.smtp_host:
log.info("email (stdout fallback): to=%s subject=%s", to_address, subject)
record_outbound(
to_address=to_address,
from_address=cfg.from_address,
subject=subject,
kind=kind,
status="deferred",
message_id=message_id,
notification_id=notification_id,
)
return True
try:
smtp = smtplib.SMTP(cfg.smtp_host, cfg.smtp_port, timeout=30)
try:
if cfg.smtp_starttls:
smtp.starttls()
if cfg.smtp_user:
smtp.login(cfg.smtp_user, cfg.smtp_password)
smtp.send_message(msg)
finally:
smtp.quit()
record_outbound(
to_address=to_address,
from_address=cfg.from_address,
subject=subject,
kind=kind,
status="sent",
message_id=message_id,
notification_id=notification_id,
)
return True
except Exception as exc:
log.exception("email send failed: to=%s subject=%s", to_address, subject)
record_outbound(
to_address=to_address,
from_address=cfg.from_address,
subject=subject,
kind=kind,
status="failed",
error=f"{type(exc).__name__}: {exc}",
message_id=message_id,
notification_id=notification_id,
)
return False
def record_outbound(
*,
to_address: str,
from_address: str,
subject: str,
kind: str,
status: str,
error: str | None = None,
notification_id: int | None = None,
message_id: str | None = None,
) -> int | None:
"""v0.18.0 Slice 4: write one row to `outbound_emails`.
Returns the inserted row's id, or `None` if the DB connection
isn't initialized (which happens in unit tests that don't boot
the full app — the write is best-effort and never raises).
"""
try:
cur = db.conn().execute(
"""
INSERT INTO outbound_emails
(to_address, from_address, subject, kind, sent_at, status,
error, notification_id, message_id)
VALUES (?, ?, ?, ?, datetime('now'), ?, ?, ?, ?)
""",
(
to_address,
from_address,
subject,
kind,
status,
error,
notification_id,
message_id,
),
)
return cur.lastrowid
except RuntimeError:
# db.conn() raises RuntimeError if init() hasn't been called.
# Pure-helper unit tests for build_envelope hit this path; the
# audit row is best-effort and not part of the contract.
return None
except Exception:
log.exception("outbound_emails write failed: to=%s subject=%s", to_address, subject)
return None
# ---------------------------------------------------------------------------
# Quiet-hours release pass — called from the digest job
# ---------------------------------------------------------------------------
def flush_pending() -> int:
"""§15.8 release-from-hold. For each user whose quiet-hours window
has ended (or who has no quiet hours configured), find notifications
whose `email_sent_at IS NULL` and whose category is enabled, and
send them. Bundle into a single "Activity while you were away"
when more than the threshold accumulated."""
cfg = EmailConfig.from_env()
if not cfg.enabled:
return 0
users = db.conn().execute(
"""
SELECT id, email, display_name, role,
email_personal_direct, email_watched_structural, email_admin_actionable,
email_opt_out_all,
notification_quiet_hours_start, notification_quiet_hours_end,
notification_quiet_hours_timezone
FROM users
"""
).fetchall()
sent_count = 0
for user in users:
if not user["email"]:
continue
if _in_quiet_hours(user):
continue
rows = db.conn().execute(
"""
SELECT n.id, n.event_kind, n.rfc_slug, n.branch_name, n.pr_number,
n.created_at, n.payload,
u.display_name AS actor_display,
r.title AS rfc_title
FROM notifications n
LEFT JOIN users u ON u.id = n.actor_user_id
LEFT JOIN cached_rfcs r ON r.slug = n.rfc_slug
WHERE n.recipient_user_id = ?
AND n.email_sent_at IS NULL
AND n.read_at IS NULL
ORDER BY n.id ASC
""",
(user["id"],),
).fetchall()
emailable = []
for r in rows:
try:
extras = json.loads(r["payload"] or "{}")
except json.JSONDecodeError:
extras = {}
cat = category_for(r["event_kind"], extras.get("category", "structural"))
if not _user_wants_email(user, cat):
continue
emailable.append((r, cat, extras))
if not emailable:
continue
if len(emailable) >= cfg.bundle_threshold:
sent_count += _send_bundle(cfg, user, emailable)
else:
for r, cat, extras in emailable:
payload = _row_to_payload(r, extras)
_send_one(user, r["id"], payload, cat)
sent_count += 1
return sent_count
def _row_to_payload(row: Any, extras: dict) -> dict:
return {
"event_kind": row["event_kind"],
"rfc_slug": row["rfc_slug"],
"rfc_title": row["rfc_title"],
"branch_name": row["branch_name"],
"pr_number": row["pr_number"],
"actor_display": row["actor_display"],
"created_at": row["created_at"],
"summary": _summary_for(row["event_kind"], row["actor_display"], row["rfc_title"], extras),
**extras,
}
def _summary_for(event_kind: str, actor_display: str | None, rfc_title: str | None, extras: dict) -> str:
# Local copy to avoid import cycle with notify.py.
from .notify import render_summary
return render_summary(event_kind, actor_display, rfc_title, extras)
def _send_bundle(cfg: EmailConfig, user: Any, emailable: list) -> int:
"""One "Activity while you were away" email per §15.4. Subject names
the count, body lists summaries grouped by RFC."""
count = len(emailable)
subject = f"[Wiggleverse] Activity while you were away — {count} events"
sections: list[str] = []
sorted_rows = sorted(emailable, key=lambda t: t[0]["rfc_slug"] or "")
for slug, group in groupby(sorted_rows, key=lambda t: t[0]["rfc_slug"]):
group_rows = list(group)
title = group_rows[0][0]["rfc_title"] or slug or "(no RFC)"
sections.append(f"\n{title}")
for r, _cat, extras in group_rows:
summary = _summary_for(r["event_kind"], r["actor_display"], r["rfc_title"], extras)
sections.append(f" · {summary}")
# v0.18.0: the bundle covers multiple categories, so a
# per-category opt-out can't honor the user's intent. The
# `all` category lands at the §15.4 endpoint and sets
# `email_opt_out_all = 1`.
unsubscribe_url = make_unsubscribe_url(user["id"], "all")
body = (
"Activity on RFCs you watch, accumulated during your quiet hours:\n"
+ "\n".join(sections)
+ f"\n\nOpen your inbox: {cfg.app_url}/inbox\n"
+ f"Manage all preferences: {cfg.app_url}/settings/notifications\n"
+ f"Unsubscribe from all email: {unsubscribe_url}\n"
)
sent = _deliver(
cfg,
user["email"],
subject,
body,
unsubscribe_mailto=cfg.unsubscribe_mailto,
unsubscribe_url=unsubscribe_url,
kind="bundle",
)
if not sent:
return 0
ids = [r["id"] for r, _, _ in emailable]
placeholders = ",".join("?" * len(ids))
db.conn().execute(
f"UPDATE notifications SET email_sent_at = datetime('now') WHERE id IN ({placeholders})",
ids,
)
return count