jack/PocketVeto
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: email unsubscribe tokens with one-click opt-out

Browse Source 
- Migration 0019: email_unsubscribe_token column on users (unique, indexed)
- Token auto-generated on first email address save (same pattern as RSS token)
- GET /api/notifications/unsubscribe/{token} — no auth required, sets
  email_enabled=False and returns a branded HTML confirmation page
- List-Unsubscribe + List-Unsubscribe-Post headers on every email
  (improves deliverability; enables one-click unsubscribe in Gmail/Outlook)
- Unsubscribe link appended to email body plain text

Authored by: Jack Levy
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Jack Levy
2026-03-14 18:56:59 -04:00
parent 49bda16ad5
commit 380ff4addb
4 changed files with 91 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
backend/app/api/notifications.py
View File

@@ -7,7 +7,7 @@ from xml.etree.ElementTree import Element, SubElement, tostring
import httpx
from fastapi import APIRouter, Depends, HTTPException
from fastapi.responses import Response
from fastapi.responses import HTMLResponse, Response
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
@@ -119,6 +119,9 @@ async def update_notification_settings(
if not user.rss_token:
user.rss_token = secrets.token_urlsafe(32)
# Generate unsubscribe token the first time an email address is saved
if prefs.get("email_address") and not user.email_unsubscribe_token:
user.email_unsubscribe_token = secrets.token_urlsafe(32)
await db.commit()
await db.refresh(user)
@@ -371,6 +374,51 @@ async def test_follow_mode(
return NotificationTestResult(status="error", detail=f"Connection error: {e}")
@router.get("/unsubscribe/{token}", response_class=HTMLResponse, include_in_schema=False)
async def email_unsubscribe(token: str, db: AsyncSession = Depends(get_db)):
"""One-click email unsubscribe — no login required."""
result = await db.execute(
select(User).where(User.email_unsubscribe_token == token)
)
user = result.scalar_one_or_none()
if not user:
return HTMLResponse(
_unsubscribe_page("Invalid or expired link", success=False),
status_code=404,
)
prefs = dict(user.notification_prefs or {})
prefs["email_enabled"] = False
user.notification_prefs = prefs
await db.commit()
return HTMLResponse(_unsubscribe_page("You've been unsubscribed from PocketVeto email notifications.", success=True))
def _unsubscribe_page(message: str, success: bool) -> str:
color = "#16a34a" if success else "#dc2626"
icon = "✓" if success else "✗"
return f"""<!DOCTYPE html>
<html lang="en">
<head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">
<title>PocketVeto — Unsubscribe</title>
<style>
body{{font-family:system-ui,sans-serif;background:#f9fafb;display:flex;align-items:center;justify-content:center;min-height:100vh;margin:0}}
.card{{background:#fff;border:1px solid #e5e7eb;border-radius:12px;padding:2.5rem;max-width:420px;width:100%;text-align:center;box-shadow:0 1px 3px rgba(0,0,0,.08)}}
.icon{{font-size:2.5rem;color:{color};margin-bottom:1rem}}
h1{{font-size:1.1rem;font-weight:600;color:#111827;margin:0 0 .5rem}}
p{{font-size:.9rem;color:#6b7280;margin:0 0 1.5rem;line-height:1.5}}
a{{color:#2563eb;text-decoration:none;font-size:.875rem}}a:hover{{text-decoration:underline}}
</style></head>
<body><div class="card">
<div class="icon">{icon}</div>
<h1>Email Notifications</h1>
<p>{message}</p>
<a href="/">Return to PocketVeto</a>
</div></body></html>"""
@router.get("/feed/{rss_token}.xml", include_in_schema=False)
async def rss_feed(rss_token: str, db: AsyncSession = Depends(get_db)):
"""Public tokenized RSS feed — no auth required."""

1
backend/app/models/user.py
View File

@@ -15,6 +15,7 @@ class User(Base):
is_admin = Column(Boolean, nullable=False, default=False)
notification_prefs = Column(JSONB, nullable=False, default=dict)
rss_token = Column(String, unique=True, nullable=True, index=True)
email_unsubscribe_token = Column(String(64), unique=True, nullable=True, index=True)
created_at = Column(DateTime(timezone=True), server_default=func.now())
follows = relationship("Follow", back_populates="user", cascade="all, delete-orphan")

22
backend/app/workers/notification_dispatcher.py
View File

@@ -192,7 +192,7 @@ def dispatch_notifications(self):
email_address = prefs.get("email_address", "").strip()
if email_enabled and email_address:
try:
_send_email(event, email_address)
_send_email(event, email_address, unsubscribe_token=user.email_unsubscribe_token)
sent += 1
except Exception as e:
logger.warning(f"email dispatch failed for event {event.id}: {e}")
@@ -285,9 +285,14 @@ def _build_reason(payload: dict) -> str | None:
return None
def _send_email(event: NotificationEvent, email_address: str) -> None:
def _send_email(
event: NotificationEvent,
email_address: str,
unsubscribe_token: str | None = None,
) -> None:
"""Send a plain-text email notification via SMTP."""
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from app.config import settings as app_settings
@@ -299,6 +304,7 @@ def _send_email(event: NotificationEvent, email_address: str) -> None:
bill_label = payload.get("bill_label", event.bill_id.upper())
bill_title = payload.get("bill_title", "")
event_label = _EVENT_TITLES.get(event.event_type, "Bill Update")
base_url = (app_settings.PUBLIC_URL or app_settings.LOCAL_URL).rstrip("/")
subject = f"PocketVeto: {event_label} — {bill_label}"
@@ -315,13 +321,23 @@ def _send_email(event: NotificationEvent, email_address: str) -> None:
if payload.get("bill_url"):
lines.append("")
lines.append(f"View bill: {payload['bill_url']}")
unsubscribe_url = f"{base_url}/api/notifications/unsubscribe/{unsubscribe_token}" if unsubscribe_token else None
if unsubscribe_url:
lines.append("")
lines.append(f"Unsubscribe from email alerts: {unsubscribe_url}")
body = "\n".join(lines)
from_addr = app_settings.SMTP_FROM or app_settings.SMTP_USER
msg = MIMEText(body, "plain", "utf-8")
msg = MIMEMultipart()
msg["Subject"] = subject
msg["From"] = from_addr
msg["To"] = email_address
if unsubscribe_url:
msg["List-Unsubscribe"] = f"<{unsubscribe_url}>"
msg["List-Unsubscribe-Post"] = "List-Unsubscribe=One-Click"
msg.attach(MIMEText(body, "plain", "utf-8"))
with smtplib.SMTP(app_settings.SMTP_HOST, app_settings.SMTP_PORT, timeout=10) as s:
if app_settings.SMTP_STARTTLS: