4 Commits
v0.2.0 ... main

Author SHA1 Message Date
Roberto Musso
df43f4783a fix(migration): guard 003 against pre-existing language column
All checks were successful
Test & Deploy Waitlist / test (push) Successful in 35s
Test & Deploy Waitlist / deploy (push) Successful in 16s
2026-04-12 10:12:52 +02:00
Roberto Musso
b7ba18641b feat(i18n): add multilanguage support to waitlist emails and result pages
Some checks failed
Test & Deploy Waitlist / test (push) Successful in 34s
Test & Deploy Waitlist / deploy (push) Failing after 15s
- Add 'language' column to waitlist_entries (en/it/es/fr/de, default en)
- Accept 'lang' field in POST /waitlist request body
- Translate confirmation email (subject, badge, heading, body, CTA, footer)
- Translate confirm/unsubscribe result HTML pages
- Return localized success message in WaitlistResponse
- Update language preference on duplicate signups
- Alembic migration 003_add_language_column
2026-04-12 10:06:35 +02:00
Roberto Musso
d32fc7ae30 feat: add daily waitlist report email (Brevo)
All checks were successful
Test & Deploy Waitlist / test (push) Successful in 35s
Test & Deploy Waitlist / deploy (push) Successful in 15s
2026-04-11 20:28:29 +02:00
Roberto Musso
352e25d651 feat: GDPR compliance — anonymization, unsubscribe, consent tracking
- Add consent_given_at and anonymized_at fields + Alembic migration (002)
- Add GET /waitlist/unsubscribe endpoint (HMAC token, anonymizes PII)
- Add cleanup.py: cron-able script to anonymize unconfirmed entries after 48h
- Clear IP address on email confirmation (no longer needed)
- Add unsubscribe link in confirmation email footer
- Record consent timestamp on signup
- Add 4 new tests (unsubscribe, consent timestamp)
- Update .env.example, schemas
2026-04-11 19:41:27 +02:00
11 changed files with 723 additions and 35 deletions

View File

@@ -20,3 +20,6 @@ BREVO_LIST_ID=0
CONFIRM_SECRET=replace-with-a-long-random-string CONFIRM_SECRET=replace-with-a-long-random-string
CONFIRM_BASE_URL=https://adiuvai.com CONFIRM_BASE_URL=https://adiuvai.com
CONFIRM_TOKEN_EXPIRY_HOURS=48 CONFIRM_TOKEN_EXPIRY_HOURS=48
# Daily report — leave empty to disable
REPORT_RECIPIENT_EMAIL=

View File

@@ -0,0 +1,25 @@
"""add consent_given_at and anonymized_at columns
Revision ID: 002
Revises: 001
Create Date: 2026-04-11
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
revision: str = "002"
down_revision: Union[str, None] = "001"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.add_column("waitlist_entries", sa.Column("consent_given_at", sa.DateTime(timezone=True), nullable=True))
op.add_column("waitlist_entries", sa.Column("anonymized_at", sa.DateTime(timezone=True), nullable=True))
def downgrade() -> None:
op.drop_column("waitlist_entries", "anonymized_at")
op.drop_column("waitlist_entries", "consent_given_at")

View File

@@ -0,0 +1,34 @@
"""add language column to waitlist_entries
Revision ID: 003
Revises: 002
Create Date: 2026-04-12
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
revision: str = "003"
down_revision: Union[str, None] = "002"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# Guard: column may already exist from a prior manual migration
op.execute("""
DO $$
BEGIN
IF NOT EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_name = 'waitlist_entries' AND column_name = 'language'
) THEN
ALTER TABLE waitlist_entries ADD COLUMN language VARCHAR(5) NOT NULL DEFAULT 'en';
END IF;
END $$;
""")
def downgrade() -> None:
op.drop_column("waitlist_entries", "language")

View File

@@ -15,6 +15,65 @@ logger = logging.getLogger(__name__)
BREVO_API = "https://api.brevo.com/v3" BREVO_API = "https://api.brevo.com/v3"
# ── Translations for transactional emails ──────────────────────────────
_EMAIL_I18N: dict[str, dict[str, str]] = {
"en": {
"subject": "Confirm your spot on the adiuvAI waitlist",
"preheader": "Confirm your email to secure your early access spot on the adiuvAI waitlist.",
"badge": "One more step",
"heading": "Confirm your spot on<br>the waitlist",
"body": "Thanks for signing up! Please confirm your email address so we can keep you in the loop when adiuvAI launches.",
"cta": "Confirm my email",
"expiry": "This link expires in {hours} hours.",
"footer": "If you didn't sign up, simply ignore this email.",
},
"it": {
"subject": "Conferma il tuo posto nella lista d'attesa di adiuvAI",
"preheader": "Conferma la tua email per assicurarti un accesso anticipato ad adiuvAI.",
"badge": "Ancora un passaggio",
"heading": "Conferma il tuo posto<br>nella lista d'attesa",
"body": "Grazie per esserti iscritto! Conferma il tuo indirizzo email così potremo aggiornarti quando adiuvAI sarà disponibile.",
"cta": "Conferma la mia email",
"expiry": "Questo link scade tra {hours} ore.",
"footer": "Se non ti sei iscritto, ignora semplicemente questa email.",
},
"es": {
"subject": "Confirma tu lugar en la lista de espera de adiuvAI",
"preheader": "Confirma tu correo para asegurar tu acceso anticipado a adiuvAI.",
"badge": "Un paso más",
"heading": "Confirma tu lugar en<br>la lista de espera",
"body": "¡Gracias por registrarte! Confirma tu dirección de correo para que podamos avisarte cuando adiuvAI esté listo.",
"cta": "Confirmar mi correo",
"expiry": "Este enlace caduca en {hours} horas.",
"footer": "Si no te registraste, simplemente ignora este correo.",
},
"fr": {
"subject": "Confirmez votre place sur la liste d'attente d'adiuvAI",
"preheader": "Confirmez votre email pour sécuriser votre accès anticipé à adiuvAI.",
"badge": "Encore une étape",
"heading": "Confirmez votre place sur<br>la liste d'attente",
"body": "Merci de vous être inscrit ! Veuillez confirmer votre adresse email pour que nous puissions vous tenir informé du lancement d'adiuvAI.",
"cta": "Confirmer mon email",
"expiry": "Ce lien expire dans {hours} heures.",
"footer": "Si vous ne vous êtes pas inscrit, ignorez simplement cet email.",
},
"de": {
"subject": "Bestätige deinen Platz auf der adiuvAI-Warteliste",
"preheader": "Bestätige deine EMail, um dir den frühen Zugang zu adiuvAI zu sichern.",
"badge": "Noch ein Schritt",
"heading": "Bestätige deinen Platz<br>auf der Warteliste",
"body": "Danke für deine Anmeldung! Bitte bestätige deine EMail-Adresse, damit wir dich informieren können, wenn adiuvAI startet.",
"cta": "Meine EMail bestätigen",
"expiry": "Dieser Link läuft in {hours} Stunden ab.",
"footer": "Falls du dich nicht angemeldet hast, ignoriere diese EMail einfach.",
},
}
def _t(lang: str, key: str) -> str:
"""Get translated string, falling back to English."""
return _EMAIL_I18N.get(lang, _EMAIL_I18N["en"]).get(key, _EMAIL_I18N["en"][key])
def _headers() -> dict[str, str]: def _headers() -> dict[str, str]:
return { return {
@@ -24,7 +83,7 @@ def _headers() -> dict[str, str]:
} }
async def send_confirmation_email(email: str, confirm_url: str) -> bool: async def send_confirmation_email(email: str, confirm_url: str, unsubscribe_url: str = "", lang: str = "en") -> bool:
"""Send a double opt-in confirmation email. Returns True on success.""" """Send a double opt-in confirmation email. Returns True on success."""
if not settings.brevo_configured: if not settings.brevo_configured:
logger.warning("Brevo not configured — skipping confirmation email for %s***", email[:3]) logger.warning("Brevo not configured — skipping confirmation email for %s***", email[:3])
@@ -36,8 +95,8 @@ async def send_confirmation_email(email: str, confirm_url: str) -> bool:
"email": settings.BREVO_SENDER_EMAIL, "email": settings.BREVO_SENDER_EMAIL,
}, },
"to": [{"email": email}], "to": [{"email": email}],
"subject": "Confirm your spot on the adiuvAI waitlist", "subject": _t(lang, "subject"),
"htmlContent": _confirmation_html(confirm_url), "htmlContent": _confirmation_html(confirm_url, unsubscribe_url, lang),
} }
try: try:
@@ -78,18 +137,27 @@ async def add_contact_to_list(email: str) -> bool:
return False return False
def _confirmation_html(confirm_url: str) -> str: def _confirmation_html(confirm_url: str, unsubscribe_url: str = "", lang: str = "en") -> str:
"""Email template aligned with the adiuvAI landing page brand.""" """Email template aligned with the adiuvAI landing page brand."""
html_lang = lang if lang in ("en", "it", "es", "fr", "de") else "en"
badge = _t(lang, "badge")
heading = _t(lang, "heading")
body = _t(lang, "body")
cta = _t(lang, "cta")
expiry = _t(lang, "expiry").format(hours=settings.CONFIRM_TOKEN_EXPIRY_HOURS)
preheader = _t(lang, "preheader")
footer = _t(lang, "footer")
unsub_label = {"en": "Unsubscribe", "it": "Annulla iscrizione", "es": "Cancelar suscripción", "fr": "Se désabonner", "de": "Abmelden"}.get(lang, "Unsubscribe")
return f"""\ return f"""\
<!DOCTYPE html> <!DOCTYPE html>
<html lang="en" xmlns="http://www.w3.org/1999/xhtml"> <html lang="{html_lang}" xmlns="http://www.w3.org/1999/xhtml">
<head> <head>
<meta charset="utf-8"> <meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1"> <meta name="viewport" content="width=device-width,initial-scale=1">
<meta http-equiv="X-UA-Compatible" content="IE=edge"> <meta http-equiv="X-UA-Compatible" content="IE=edge">
<meta name="color-scheme" content="light"> <meta name="color-scheme" content="light">
<meta name="supported-color-schemes" content="light"> <meta name="supported-color-schemes" content="light">
<title>Confirm your spot — adiuvAI</title> <title>{_t(lang, "subject")} — adiuvAI</title>
<!--[if mso]><style>table,td{{font-family:Arial,sans-serif!important;}}</style><![endif]--> <!--[if mso]><style>table,td{{font-family:Arial,sans-serif!important;}}</style><![endif]-->
</head> </head>
<body style="margin:0;padding:0;width:100%;background-color:#f4edf3; <body style="margin:0;padding:0;width:100%;background-color:#f4edf3;
@@ -98,7 +166,7 @@ def _confirmation_html(confirm_url: str) -> str:
<!-- Preheader (hidden text for inbox preview) --> <!-- Preheader (hidden text for inbox preview) -->
<div style="display:none;font-size:1px;color:#f4edf3;line-height:1px;max-height:0;max-width:0;opacity:0;overflow:hidden;"> <div style="display:none;font-size:1px;color:#f4edf3;line-height:1px;max-height:0;max-width:0;opacity:0;overflow:hidden;">
Confirm your email to secure your early access spot on the adiuvAI waitlist. {preheader}
</div> </div>
<!-- Outer wrapper --> <!-- Outer wrapper -->
@@ -156,21 +224,20 @@ def _confirmation_html(confirm_url: str) -> str:
border:1px solid rgba(251,200,129,0.2); border:1px solid rgba(251,200,129,0.2);
font-size:11px;font-weight:600;letter-spacing:0.06em; font-size:11px;font-weight:600;letter-spacing:0.06em;
text-transform:uppercase;color:#e5a94e;"> text-transform:uppercase;color:#e5a94e;">
&#9679;&ensp;One more step &#9679;&ensp;{badge}
</td></tr> </td></tr>
</table> </table>
<!-- Heading --> <!-- Heading -->
<h1 style="margin:0 0 16px;font-size:24px;font-weight:600; <h1 style="margin:0 0 16px;font-size:24px;font-weight:600;
letter-spacing:-0.03em;line-height:1.2;color:#040404;text-align:center;"> letter-spacing:-0.03em;line-height:1.2;color:#040404;text-align:center;">
Confirm your spot on<br>the waitlist {heading}
</h1> </h1>
<!-- Paragraph --> <!-- Paragraph -->
<p style="margin:0 0 28px;font-size:15px;line-height:1.7;color:#8a8ea9; <p style="margin:0 0 28px;font-size:15px;line-height:1.7;color:#8a8ea9;
text-align:center;"> text-align:center;">
Thanks for signing up! Please confirm your email address {body}
so we can keep you in the loop when adiuvAI launches.
</p> </p>
<!-- CTA Button --> <!-- CTA Button -->
@@ -184,14 +251,14 @@ def _confirmation_html(confirm_url: str) -> str:
color:#fbfbfb;text-decoration:none; color:#fbfbfb;text-decoration:none;
border-radius:50px; border-radius:50px;
font-family:-apple-system,BlinkMacSystemFont,'Segoe UI',Roboto,sans-serif;"> font-family:-apple-system,BlinkMacSystemFont,'Segoe UI',Roboto,sans-serif;">
Confirm my email {cta}
</a> </a>
</td></tr> </td></tr>
</table> </table>
<!-- Expiry note --> <!-- Expiry note -->
<p style="margin:20px 0 0;font-size:13px;color:#c8c3cd;text-align:center;"> <p style="margin:20px 0 0;font-size:13px;color:#c8c3cd;text-align:center;">
This link expires in {settings.CONFIRM_TOKEN_EXPIRY_HOURS} hours. {expiry}
</p> </p>
</td></tr> </td></tr>
@@ -206,11 +273,12 @@ def _confirmation_html(confirm_url: str) -> str:
<!-- Footer --> <!-- Footer -->
<tr><td style="padding:20px 40px 32px;text-align:center;"> <tr><td style="padding:20px 40px 32px;text-align:center;">
<p style="margin:0;font-size:12px;color:#c8c3cd;line-height:1.5;"> <p style="margin:0;font-size:12px;color:#c8c3cd;line-height:1.5;">
If you didn't sign up, simply ignore this email. {footer}
</p> </p>
<p style="margin:8px 0 0;font-size:12px;color:#c8c3cd;"> <p style="margin:8px 0 0;font-size:12px;color:#c8c3cd;">
<a href="https://adiuvai.com" style="color:#8a8ea9;text-decoration:underline; <a href="https://adiuvai.com" style="color:#8a8ea9;text-decoration:underline;
text-underline-offset:2px;">adiuvai.com</a> text-underline-offset:2px;">adiuvai.com</a>
{f'&ensp;·&ensp;<a href="{unsubscribe_url}" style="color:#8a8ea9;text-decoration:underline;text-underline-offset:2px;">{unsub_label}</a>' if unsubscribe_url else ''}
</p> </p>
</td></tr> </td></tr>

58
app/cleanup.py Normal file
View File

@@ -0,0 +1,58 @@
"""
Periodic cleanup: anonymize unconfirmed waitlist entries older than CONFIRM_TOKEN_EXPIRY_HOURS.
Run as a cron job:
python -m app.cleanup
Or integrate into docker-compose with a one-shot service.
"""
import asyncio
import datetime
import logging
from sqlalchemy import select, and_
from app.config import settings
from app.db import async_session
from app.models import WaitlistEntry
from app.routes import _anonymize_entry
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
async def anonymize_expired() -> int:
"""Anonymize all unconfirmed entries past the token expiry window. Returns count."""
cutoff = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(
hours=settings.CONFIRM_TOKEN_EXPIRY_HOURS
)
async with async_session() as db:
result = await db.execute(
select(WaitlistEntry).where(
and_(
WaitlistEntry.confirmed == False, # noqa: E712
WaitlistEntry.anonymized_at == None, # noqa: E711
WaitlistEntry.created_at < cutoff,
)
)
)
entries = result.scalars().all()
for entry in entries:
_anonymize_entry(entry)
await db.commit()
if entries:
logger.info("Anonymized %d expired unconfirmed entries", len(entries))
else:
logger.info("No expired unconfirmed entries to anonymize")
return len(entries)
if __name__ == "__main__":
count = asyncio.run(anonymize_expired())
raise SystemExit(0)

View File

@@ -20,6 +20,9 @@ class Settings(BaseSettings):
CONFIRM_BASE_URL: str = "https://adiuvai.com" CONFIRM_BASE_URL: str = "https://adiuvai.com"
CONFIRM_TOKEN_EXPIRY_HOURS: int = 48 CONFIRM_TOKEN_EXPIRY_HOURS: int = 48
# Daily report
REPORT_RECIPIENT_EMAIL: str = "" # your email — leave empty to disable
model_config = {"env_file": ".env", "env_file_encoding": "utf-8"} model_config = {"env_file": ".env", "env_file_encoding": "utf-8"}
@property @property

276
app/daily_report.py Normal file
View File

@@ -0,0 +1,276 @@
"""
Daily waitlist report — sends an evening summary email via Brevo.
Run as a cron job (e.g. every day at 21:00):
python -m app.daily_report
Requires REPORT_RECIPIENT_EMAIL and BREVO_API_KEY in .env.
"""
import asyncio
import datetime
import logging
import httpx
from sqlalchemy import func, select, and_
from app.config import settings
from app.db import async_session
from app.models import WaitlistEntry
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
async def gather_stats() -> dict:
"""Collect today's waitlist statistics."""
now = datetime.datetime.now(datetime.timezone.utc)
today_start = now.replace(hour=0, minute=0, second=0, microsecond=0)
async with async_session() as db:
# Total counts
total = (await db.execute(select(func.count(WaitlistEntry.id)))).scalar() or 0
confirmed_total = (await db.execute(
select(func.count(WaitlistEntry.id)).where(WaitlistEntry.confirmed == True) # noqa: E712
)).scalar() or 0
anonymized_total = (await db.execute(
select(func.count(WaitlistEntry.id)).where(WaitlistEntry.anonymized_at != None) # noqa: E711
)).scalar() or 0
pending = total - confirmed_total - anonymized_total
# Today's activity
new_today = (await db.execute(
select(func.count(WaitlistEntry.id)).where(
and_(
WaitlistEntry.created_at >= today_start,
WaitlistEntry.anonymized_at == None, # noqa: E711
)
)
)).scalar() or 0
confirmed_today = (await db.execute(
select(func.count(WaitlistEntry.id)).where(
and_(
WaitlistEntry.confirmed == True, # noqa: E712
WaitlistEntry.consent_given_at >= today_start,
)
)
)).scalar() or 0
anonymized_today = (await db.execute(
select(func.count(WaitlistEntry.id)).where(
and_(
WaitlistEntry.anonymized_at != None, # noqa: E711
WaitlistEntry.anonymized_at >= today_start,
)
)
)).scalar() or 0
return {
"date": now.strftime("%B %d, %Y"),
"total": total,
"confirmed_total": confirmed_total,
"pending": pending,
"anonymized_total": anonymized_total,
"new_today": new_today,
"confirmed_today": confirmed_today,
"anonymized_today": anonymized_today,
"conversion_rate": round(confirmed_total / total * 100, 1) if total > 0 else 0,
}
async def send_report() -> bool:
"""Gather stats and send the daily report email."""
if not settings.REPORT_RECIPIENT_EMAIL:
logger.warning("REPORT_RECIPIENT_EMAIL not set — skipping daily report")
return False
if not settings.brevo_configured:
logger.warning("Brevo not configured — skipping daily report")
return False
stats = await gather_stats()
html = _report_html(stats)
payload = {
"sender": {
"name": settings.BREVO_SENDER_NAME,
"email": settings.BREVO_SENDER_EMAIL,
},
"to": [{"email": settings.REPORT_RECIPIENT_EMAIL}],
"subject": f"Waitlist report — {stats['date']}",
"htmlContent": html,
}
headers = {
"api-key": settings.BREVO_API_KEY,
"Content-Type": "application/json",
"Accept": "application/json",
}
try:
async with httpx.AsyncClient(timeout=10) as client:
resp = await client.post(
"https://api.brevo.com/v3/smtp/email",
headers=headers,
json=payload,
)
resp.raise_for_status()
logger.info("Daily report sent to %s", settings.REPORT_RECIPIENT_EMAIL)
return True
except httpx.HTTPError:
logger.exception("Failed to send daily report")
return False
def _report_html(s: dict) -> str:
"""adiuvAI-branded daily report email."""
def _stat_cell(label: str, value, color: str = "#040404") -> str:
return f"""\
<td align="center" style="padding:16px 8px;">
<p style="margin:0;font-size:28px;font-weight:700;color:{color};
letter-spacing:-0.03em;line-height:1;">{value}</p>
<p style="margin:6px 0 0;font-size:11px;font-weight:600;color:#8a8ea9;
text-transform:uppercase;letter-spacing:0.06em;">{label}</p>
</td>"""
def _row(label: str, value, highlight: bool = False) -> str:
color = "#e5a94e" if highlight else "#040404"
return f"""\
<tr>
<td style="padding:10px 0;font-size:14px;color:#8a8ea9;border-bottom:1px solid #f4edf3;">
{label}
</td>
<td align="right" style="padding:10px 0;font-size:14px;font-weight:600;color:{color};
border-bottom:1px solid #f4edf3;">
{value}
</td>
</tr>"""
return f"""\
<!DOCTYPE html>
<html lang="en" xmlns="http://www.w3.org/1999/xhtml">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1">
<meta name="color-scheme" content="light">
<title>Waitlist Report — {s['date']}</title>
<!--[if mso]><style>table,td{{font-family:Arial,sans-serif!important;}}</style><![endif]-->
</head>
<body style="margin:0;padding:0;width:100%;background-color:#f4edf3;
font-family:-apple-system,BlinkMacSystemFont,'Segoe UI',Roboto,'Helvetica Neue',Arial,sans-serif;
-webkit-font-smoothing:antialiased;">
<div style="display:none;font-size:1px;color:#f4edf3;line-height:1px;max-height:0;max-width:0;opacity:0;overflow:hidden;">
{s['new_today']} new signups, {s['confirmed_total']} confirmed total ({s['conversion_rate']}% conversion)
</div>
<table role="presentation" width="100%" cellpadding="0" cellspacing="0" border="0"
style="background-color:#f4edf3;">
<tr><td align="center" style="padding:48px 16px;">
<table role="presentation" width="520" cellpadding="0" cellspacing="0" border="0"
style="max-width:520px;width:100%;background-color:#ffffff;
border-radius:20px;overflow:hidden;
box-shadow:0 4px 16px rgba(0,0,0,0.03),0 1px 2px rgba(0,0,0,0.02);">
<!-- Gold bar -->
<tr><td style="height:4px;background:linear-gradient(135deg,#fbc881 0%,#e5a94e 100%);
font-size:0;line-height:0;">&nbsp;</td></tr>
<!-- Logo -->
<tr><td align="center" style="padding:32px 40px 0;">
<table role="presentation" cellpadding="0" cellspacing="0" border="0">
<tr>
<td style="padding-right:8px;vertical-align:middle;">
<!--[if !mso]><!-->
<svg viewBox="0 0 64 64" width="24" height="24" xmlns="http://www.w3.org/2000/svg"
style="display:block;width:24px;height:24px;">
<path d="M32,4 L48,32 L16,32 Z" fill="#fbc881"/>
<path d="M16,32 L48,32 L32,60 Z" fill="#040404"/>
<circle cx="32" cy="32" r="2.5" fill="#040404" opacity="0.18"/>
</svg>
<!--<![endif]-->
</td>
<td style="vertical-align:middle;font-size:16px;font-weight:400;
color:#040404;letter-spacing:-0.02em;">
adiuv<span style="font-weight:700;color:#e5a94e;">AI</span>
</td>
</tr>
</table>
</td></tr>
<!-- Header -->
<tr><td style="padding:24px 40px 0;text-align:center;">
<p style="margin:0 0 4px;font-size:11px;font-weight:600;letter-spacing:0.06em;
text-transform:uppercase;color:#e5a94e;">Daily Report</p>
<h1 style="margin:0;font-size:22px;font-weight:600;color:#040404;
letter-spacing:-0.03em;">{s['date']}</h1>
</td></tr>
<!-- Hero stats -->
<tr><td style="padding:28px 40px 0;">
<table role="presentation" width="100%" cellpadding="0" cellspacing="0" border="0"
style="background:#f4edf3;border-radius:14px;">
<tr>
{_stat_cell("Confirmed", s['confirmed_total'], "#040404")}
{_stat_cell("Pending", s['pending'], "#8a8ea9")}
{_stat_cell("Conversion", f"{s['conversion_rate']}%", "#e5a94e")}
</tr>
</table>
</td></tr>
<!-- Today's activity -->
<tr><td style="padding:28px 40px 0;">
<p style="margin:0 0 12px;font-size:11px;font-weight:600;letter-spacing:0.06em;
text-transform:uppercase;color:#e5a94e;">Today's Activity</p>
<table role="presentation" width="100%" cellpadding="0" cellspacing="0" border="0">
{_row("New signups", f"+{s['new_today']}", highlight=s['new_today'] > 0)}
{_row("Confirmed", f"+{s['confirmed_today']}", highlight=s['confirmed_today'] > 0)}
{_row("Anonymized (expired/unsub)", s['anonymized_today'])}
</table>
</td></tr>
<!-- Totals -->
<tr><td style="padding:24px 40px 0;">
<p style="margin:0 0 12px;font-size:11px;font-weight:600;letter-spacing:0.06em;
text-transform:uppercase;color:#e5a94e;">All Time</p>
<table role="presentation" width="100%" cellpadding="0" cellspacing="0" border="0">
{_row("Total records", s['total'])}
{_row("Confirmed", s['confirmed_total'], highlight=True)}
{_row("Pending confirmation", s['pending'])}
{_row("Anonymized", s['anonymized_total'])}
</table>
</td></tr>
<!-- Divider -->
<tr><td style="padding:28px 40px 0;">
<table role="presentation" width="100%" cellpadding="0" cellspacing="0" border="0">
<tr><td style="height:1px;background-color:#f4edf3;font-size:0;line-height:0;">&nbsp;</td></tr>
</table>
</td></tr>
<!-- Footer -->
<tr><td style="padding:16px 40px 28px;text-align:center;">
<p style="margin:0;font-size:12px;color:#c8c3cd;">
Automated report from
<a href="https://adiuvai.com" style="color:#8a8ea9;text-decoration:underline;
text-underline-offset:2px;">adiuvai.com</a>
</p>
</td></tr>
</table>
</td></tr>
</table>
</body>
</html>"""
if __name__ == "__main__":
asyncio.run(send_report())

View File

@@ -20,7 +20,14 @@ class WaitlistEntry(Base):
email: Mapped[str] = mapped_column(String(320), unique=True, nullable=False, index=True) email: Mapped[str] = mapped_column(String(320), unique=True, nullable=False, index=True)
ip_address: Mapped[str | None] = mapped_column(String(45), nullable=True) ip_address: Mapped[str | None] = mapped_column(String(45), nullable=True)
source: Mapped[str | None] = mapped_column(String(64), nullable=True) source: Mapped[str | None] = mapped_column(String(64), nullable=True)
language: Mapped[str] = mapped_column(String(5), nullable=False, server_default=sa.text("'en'"))
confirmed: Mapped[bool] = mapped_column(Boolean, default=False, server_default=sa.text("0")) confirmed: Mapped[bool] = mapped_column(Boolean, default=False, server_default=sa.text("0"))
consent_given_at: Mapped[datetime.datetime | None] = mapped_column(
DateTime(timezone=True), nullable=True,
)
anonymized_at: Mapped[datetime.datetime | None] = mapped_column(
DateTime(timezone=True), nullable=True,
)
created_at: Mapped[datetime.datetime] = mapped_column( created_at: Mapped[datetime.datetime] = mapped_column(
DateTime(timezone=True), DateTime(timezone=True),
server_default=sa.text("CURRENT_TIMESTAMP"), server_default=sa.text("CURRENT_TIMESTAMP"),

View File

@@ -1,4 +1,5 @@
import asyncio import asyncio
import datetime
import logging import logging
from fastapi import APIRouter, Depends, Request from fastapi import APIRouter, Depends, Request
@@ -31,13 +32,16 @@ async def join_waitlist(
- Duplicate emails: idempotent — returns success without error. - Duplicate emails: idempotent — returns success without error.
- Stores the Cloudflare-resolved client IP for analytics (not exposed). - Stores the Cloudflare-resolved client IP for analytics (not exposed).
- Sends a confirmation email via Brevo (fire-and-forget). - Sends a confirmation email via Brevo (fire-and-forget).
- Records consent timestamp (GDPR Art. 7).
""" """
# Honeypot — bots fill hidden fields; silently "succeed" # Honeypot — bots fill hidden fields; silently "succeed"
if body.website: if body.website:
return WaitlistResponse() return WaitlistResponse.for_lang(body.lang)
email = body.email.lower().strip() email = body.email.lower().strip()
lang = body.lang
ip = _get_client_ip(request) ip = _get_client_ip(request)
now = datetime.datetime.now(datetime.timezone.utc)
# Check for existing entry — idempotent # Check for existing entry — idempotent
existing = await db.execute( existing = await db.execute(
@@ -46,14 +50,25 @@ async def join_waitlist(
entry = existing.scalar_one_or_none() entry = existing.scalar_one_or_none()
if entry is not None: if entry is not None:
# Re-send confirmation if not yet confirmed # Update language preference if changed
if not entry.confirmed and settings.brevo_configured: if entry.language != lang:
entry.language = lang
await db.commit()
# Re-send confirmation if not yet confirmed (and not anonymized)
if not entry.confirmed and not entry.anonymized_at and settings.brevo_configured:
token = generate_token(email) token = generate_token(email)
confirm_url = f"{settings.CONFIRM_BASE_URL}/api/v1/waitlist/confirm?token={token}" confirm_url = f"{settings.CONFIRM_BASE_URL}/api/v1/waitlist/confirm?token={token}"
asyncio.create_task(send_confirmation_email(email, confirm_url)) unsub_url = f"{settings.CONFIRM_BASE_URL}/api/v1/waitlist/unsubscribe?token={token}"
return WaitlistResponse() asyncio.create_task(send_confirmation_email(email, confirm_url, unsub_url, lang=lang))
return WaitlistResponse.for_lang(lang)
entry = WaitlistEntry(email=email, ip_address=ip, source="website") entry = WaitlistEntry(
email=email,
ip_address=ip,
source="website",
language=lang,
consent_given_at=now,
)
db.add(entry) db.add(entry)
await db.commit() await db.commit()
@@ -63,9 +78,10 @@ async def join_waitlist(
if settings.brevo_configured: if settings.brevo_configured:
token = generate_token(email) token = generate_token(email)
confirm_url = f"{settings.CONFIRM_BASE_URL}/api/v1/waitlist/confirm?token={token}" confirm_url = f"{settings.CONFIRM_BASE_URL}/api/v1/waitlist/confirm?token={token}"
asyncio.create_task(send_confirmation_email(email, confirm_url)) unsub_url = f"{settings.CONFIRM_BASE_URL}/api/v1/waitlist/unsubscribe?token={token}"
asyncio.create_task(send_confirmation_email(email, confirm_url, unsub_url, lang=lang))
return WaitlistResponse() return WaitlistResponse.for_lang(lang)
@router.get("/waitlist/confirm", response_class=HTMLResponse) @router.get("/waitlist/confirm", response_class=HTMLResponse)
@@ -90,8 +106,12 @@ async def confirm_email(
if entry is None: if entry is None:
return HTMLResponse(content=_result_page(success=False), status_code=400) return HTMLResponse(content=_result_page(success=False), status_code=400)
lang = entry.language or "en"
if not entry.confirmed: if not entry.confirmed:
entry.confirmed = True entry.confirmed = True
# Clear IP now that consent is confirmed — no longer needed
entry.ip_address = None
await db.commit() await db.commit()
logger.info("Email confirmed: %s***", email[:3]) logger.info("Email confirmed: %s***", email[:3])
@@ -99,27 +119,140 @@ async def confirm_email(
if settings.brevo_configured: if settings.brevo_configured:
asyncio.create_task(add_contact_to_list(email)) asyncio.create_task(add_contact_to_list(email))
return HTMLResponse(content=_result_page(success=True)) return HTMLResponse(content=_result_page(success=True, lang=lang))
def _result_page(*, success: bool) -> str: @router.get("/waitlist/unsubscribe", response_class=HTMLResponse)
"""Branded HTML response for confirmation result, matching the adiuvAI landing page.""" async def unsubscribe(
token: str,
db: AsyncSession = Depends(get_db),
) -> HTMLResponse:
"""
GDPR erasure (Art. 17) — anonymize the entry.
Uses the same HMAC token system as confirmation.
"""
email = verify_token(token)
if email is None:
return HTMLResponse(
content=_result_page(success=False, variant="unsubscribe"),
status_code=400,
)
result = await db.execute(
select(WaitlistEntry).where(WaitlistEntry.email == email)
)
entry = result.scalar_one_or_none()
if entry is None:
# Already gone — show success anyway (idempotent)
return HTMLResponse(content=_result_page(success=True, variant="unsubscribe"))
lang = entry.language or "en"
_anonymize_entry(entry)
await db.commit()
logger.info("Waitlist entry anonymized (unsubscribe)")
return HTMLResponse(content=_result_page(success=True, variant="unsubscribe", lang=lang))
def _anonymize_entry(entry: WaitlistEntry) -> None:
"""Strip all PII from a waitlist entry, keeping only anonymous analytics."""
now = datetime.datetime.now(datetime.timezone.utc)
entry.email = f"anon-{entry.id}@removed.invalid"
entry.ip_address = None
entry.consent_given_at = None
entry.anonymized_at = now
def _result_page(*, success: bool, variant: str = "confirm", lang: str = "en") -> str:
"""Branded HTML response for confirmation/unsubscribe result."""
# ── Translations for result pages ──
_PAGE_I18N: dict[str, dict[str, str]] = {
"en": {
"confirmed_title": "You&#8217;re confirmed!",
"confirmed_msg": "Your email has been verified. We&#8217;ll notify you when adiuvAI is ready.",
"invalid_confirm": "This confirmation link is no longer valid. Please sign up again.",
"removed_title": "Data removed",
"removed_msg": "Your personal data has been anonymized. You will not receive any further emails from us.",
"invalid_unsub": "This unsubscribe link is no longer valid. Contact privacy@adiuvai.com if you need help.",
"invalid_title": "Invalid or expired link",
"btn": "Go to adiuvai.com",
},
"it": {
"confirmed_title": "Sei confermato!",
"confirmed_msg": "La tua email è stata verificata. Ti avviseremo quando adiuvAI sarà pronto.",
"invalid_confirm": "Questo link di conferma non è più valido. Iscriviti di nuovo.",
"removed_title": "Dati rimossi",
"removed_msg": "I tuoi dati personali sono stati anonimizzati. Non riceverai più email da noi.",
"invalid_unsub": "Questo link di cancellazione non è più valido. Contatta privacy@adiuvai.com per assistenza.",
"invalid_title": "Link non valido o scaduto",
"btn": "Vai su adiuvai.com",
},
"es": {
"confirmed_title": "¡Estás confirmado!",
"confirmed_msg": "Tu correo ha sido verificado. Te avisaremos cuando adiuvAI esté listo.",
"invalid_confirm": "Este enlace de confirmación ya no es válido. Regístrate de nuevo.",
"removed_title": "Datos eliminados",
"removed_msg": "Tus datos personales han sido anonimizados. No recibirás más correos de nuestra parte.",
"invalid_unsub": "Este enlace de cancelación ya no es válido. Contacta privacy@adiuvai.com si necesitas ayuda.",
"invalid_title": "Enlace no válido o expirado",
"btn": "Ir a adiuvai.com",
},
"fr": {
"confirmed_title": "Vous êtes confirmé !",
"confirmed_msg": "Votre email a été vérifié. Nous vous préviendrons quand adiuvAI sera prêt.",
"invalid_confirm": "Ce lien de confirmation n'est plus valide. Veuillez vous inscrire à nouveau.",
"removed_title": "Données supprimées",
"removed_msg": "Vos données personnelles ont été anonymisées. Vous ne recevrez plus d'emails de notre part.",
"invalid_unsub": "Ce lien de désinscription n'est plus valide. Contactez privacy@adiuvai.com pour obtenir de l'aide.",
"invalid_title": "Lien non valide ou expiré",
"btn": "Aller sur adiuvai.com",
},
"de": {
"confirmed_title": "Du bist bestätigt!",
"confirmed_msg": "Deine EMail wurde verifiziert. Wir benachrichtigen dich, wenn adiuvAI bereit ist.",
"invalid_confirm": "Dieser Bestätigungslink ist nicht mehr gültig. Bitte melde dich erneut an.",
"removed_title": "Daten entfernt",
"removed_msg": "Deine persönlichen Daten wurden anonymisiert. Du wirst keine weiteren EMails von uns erhalten.",
"invalid_unsub": "Dieser Abmeldelink ist nicht mehr gültig. Kontaktiere privacy@adiuvai.com für Hilfe.",
"invalid_title": "Ungültiger oder abgelaufener Link",
"btn": "Zu adiuvai.com",
},
}
t = _PAGE_I18N.get(lang, _PAGE_I18N["en"])
html_lang = lang if lang in _PAGE_I18N else "en"
if variant == "unsubscribe":
if success: if success:
title = "You&#8217;re confirmed!" title = t["removed_title"]
message = "Your email has been verified. We&#8217;ll notify you when adiuvAI is ready." message = t["removed_msg"]
icon_bg = "rgba(251,200,129,0.12)" icon_bg = "rgba(251,200,129,0.12)"
icon_border = "rgba(251,200,129,0.25)" icon_border = "rgba(251,200,129,0.25)"
icon_svg = '<svg viewBox="0 0 24 24" width="24" height="24" fill="none" stroke="#e5a94e" stroke-width="2.5" stroke-linecap="round" stroke-linejoin="round"><polyline points="20 6 9 17 4 12"/></svg>' icon_svg = '<svg viewBox="0 0 24 24" width="24" height="24" fill="none" stroke="#e5a94e" stroke-width="2.5" stroke-linecap="round" stroke-linejoin="round"><polyline points="20 6 9 17 4 12"/></svg>'
else: else:
title = "Invalid or expired link" title = t["invalid_title"]
message = "This confirmation link is no longer valid. Please sign up again." message = t["invalid_unsub"]
icon_bg = "rgba(220,38,38,0.08)"
icon_border = "rgba(220,38,38,0.18)"
icon_svg = '<svg viewBox="0 0 24 24" width="24" height="24" fill="none" stroke="#dc2626" stroke-width="2.5" stroke-linecap="round" stroke-linejoin="round"><line x1="18" y1="6" x2="6" y2="18"/><line x1="6" y1="6" x2="18" y2="18"/></svg>'
elif success:
title = t["confirmed_title"]
message = t["confirmed_msg"]
icon_bg = "rgba(251,200,129,0.12)"
icon_border = "rgba(251,200,129,0.25)"
icon_svg = '<svg viewBox="0 0 24 24" width="24" height="24" fill="none" stroke="#e5a94e" stroke-width="2.5" stroke-linecap="round" stroke-linejoin="round"><polyline points="20 6 9 17 4 12"/></svg>'
else:
title = t["invalid_title"]
message = t["invalid_confirm"]
icon_bg = "rgba(220,38,38,0.08)" icon_bg = "rgba(220,38,38,0.08)"
icon_border = "rgba(220,38,38,0.18)" icon_border = "rgba(220,38,38,0.18)"
icon_svg = '<svg viewBox="0 0 24 24" width="24" height="24" fill="none" stroke="#dc2626" stroke-width="2.5" stroke-linecap="round" stroke-linejoin="round"><line x1="18" y1="6" x2="6" y2="18"/><line x1="6" y1="6" x2="18" y2="18"/></svg>' icon_svg = '<svg viewBox="0 0 24 24" width="24" height="24" fill="none" stroke="#dc2626" stroke-width="2.5" stroke-linecap="round" stroke-linejoin="round"><line x1="18" y1="6" x2="6" y2="18"/><line x1="6" y1="6" x2="18" y2="18"/></svg>'
btn_label = t["btn"]
return f"""\ return f"""\
<!DOCTYPE html> <!DOCTYPE html>
<html lang="en"> <html lang="{html_lang}">
<head> <head>
<meta charset="utf-8"> <meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1"> <meta name="viewport" content="width=device-width,initial-scale=1">
@@ -199,7 +332,7 @@ def _result_page(*, success: bool) -> str:
<div class="icon">{icon_svg}</div> <div class="icon">{icon_svg}</div>
<h1>{title}</h1> <h1>{title}</h1>
<p>{message}</p> <p>{message}</p>
<a href="https://adiuvai.com" class="btn">Go to adiuvai.com</a> <a href="https://adiuvai.com" class="btn">{btn_label}</a>
</div> </div>
</body> </body>
</html>""" </html>"""

View File

@@ -1,12 +1,30 @@
from typing import Literal
from pydantic import BaseModel, EmailStr, Field from pydantic import BaseModel, EmailStr, Field
SUPPORTED_LANGS = ("en", "it", "es", "fr", "de")
class WaitlistRequest(BaseModel): class WaitlistRequest(BaseModel):
email: EmailStr email: EmailStr
lang: Literal["en", "it", "es", "fr", "de"] = "en"
# Honeypot field — must be empty. Bots tend to fill hidden fields. # Honeypot field — must be empty. Bots tend to fill hidden fields.
website: str = Field(default="", max_length=0) website: str = Field(default="", max_length=0)
_RESPONSE_MESSAGES: dict[str, str] = {
"en": "Check your inbox for a confirmation link!",
"it": "Controlla la tua casella email per il link di conferma!",
"es": "¡Revisa tu bandeja de entrada para el enlace de confirmación!",
"fr": "Vérifiez votre boîte de réception pour le lien de confirmation !",
"de": "Überprüfe deinen Posteingang für den Bestätigungslink!",
}
class WaitlistResponse(BaseModel): class WaitlistResponse(BaseModel):
ok: bool = True ok: bool = True
message: str = "Check your inbox for a confirmation link!" message: str = "Check your inbox for a confirmation link!"
@classmethod
def for_lang(cls, lang: str = "en") -> "WaitlistResponse":
return cls(message=_RESPONSE_MESSAGES.get(lang, _RESPONSE_MESSAGES["en"]))

View File

@@ -153,9 +153,9 @@ async def test_token_tampered():
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_confirm_valid_token(client, db_session): async def test_confirm_valid_token(client, db_session):
"""GET /confirm with valid token marks email as confirmed.""" """GET /confirm with valid token marks email as confirmed and clears IP."""
# Seed an unconfirmed entry # Seed an unconfirmed entry
entry = WaitlistEntry(email="confirm@example.com", source="website") entry = WaitlistEntry(email="confirm@example.com", source="website", ip_address="1.2.3.4")
db_session.add(entry) db_session.add(entry)
await db_session.commit() await db_session.commit()
@@ -168,7 +168,9 @@ async def test_confirm_valid_token(client, db_session):
result = await db_session.execute( result = await db_session.execute(
select(WaitlistEntry).where(WaitlistEntry.email == "confirm@example.com") select(WaitlistEntry).where(WaitlistEntry.email == "confirm@example.com")
) )
assert result.scalar_one().confirmed is True confirmed_entry = result.scalar_one()
assert confirmed_entry.confirmed is True
assert confirmed_entry.ip_address is None # IP cleared on confirm
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -224,3 +226,64 @@ async def test_signup_triggers_confirmation_email(client, db_session):
call_args = mock_send.call_args call_args = mock_send.call_args
assert call_args[0][0] == "brevo@example.com" assert call_args[0][0] == "brevo@example.com"
assert "confirm" in call_args[0][1] assert "confirm" in call_args[0][1]
# ── Unsubscribe / anonymize tests ────────────────────────────────────
@pytest.mark.asyncio
async def test_unsubscribe_anonymizes_entry(client, db_session):
"""GET /unsubscribe with valid token anonymizes the entry."""
entry = WaitlistEntry(email="unsub@example.com", source="website", confirmed=True)
db_session.add(entry)
await db_session.commit()
entry_id = entry.id
token = generate_token("unsub@example.com")
resp = await client.get(f"/api/v1/waitlist/unsubscribe?token={token}")
assert resp.status_code == 200
assert "removed" in resp.text.lower() or "anonymized" in resp.text.lower()
# Verify anonymization
result = await db_session.execute(
select(WaitlistEntry).where(WaitlistEntry.id == entry_id)
)
anon = result.scalar_one()
assert anon.email == f"anon-{entry_id}@removed.invalid"
assert anon.ip_address is None
assert anon.consent_given_at is None
assert anon.anonymized_at is not None
@pytest.mark.asyncio
async def test_unsubscribe_invalid_token(client):
"""GET /unsubscribe with invalid token returns 400."""
resp = await client.get("/api/v1/waitlist/unsubscribe?token=garbage")
assert resp.status_code == 400
@pytest.mark.asyncio
async def test_unsubscribe_already_gone(client):
"""GET /unsubscribe for non-existent entry returns 200 (idempotent)."""
token = generate_token("gone@example.com")
resp = await client.get(f"/api/v1/waitlist/unsubscribe?token={token}")
assert resp.status_code == 200
# ── Consent timestamp tests ──────────────────────────────────────────
@pytest.mark.asyncio
async def test_signup_records_consent_timestamp(client, db_session):
"""New signup records consent_given_at."""
resp = await client.post(
"/api/v1/waitlist",
json={"email": "consent@example.com"},
)
assert resp.status_code == 200
result = await db_session.execute(
select(WaitlistEntry).where(WaitlistEntry.email == "consent@example.com")
)
entry = result.scalar_one()
assert entry.consent_given_at is not None