6 Commits

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
Roberto Musso
5f79ce87f9 fix: update confirm base URL to adiuvai.com and success messages for double opt-in
All checks were successful
Test & Deploy Waitlist / test (push) Successful in 35s
Test & Deploy Waitlist / deploy (push) Successful in 16s
2026-04-11 19:13:59 +02:00
Roberto Musso
73a76c6667 fix: remove unused urlencode import
All checks were successful
Test & Deploy Waitlist / test (push) Successful in 35s
Test & Deploy Waitlist / deploy (push) Successful in 37s
2026-04-11 19:00:38 +02:00
11 changed files with 727 additions and 40 deletions

View File

@@ -18,5 +18,8 @@ BREVO_LIST_ID=0
# Confirmation link # Confirmation link
CONFIRM_SECRET=replace-with-a-long-random-string CONFIRM_SECRET=replace-with-a-long-random-string
CONFIRM_BASE_URL=https://waitlist.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

@@ -17,9 +17,12 @@ class Settings(BaseSettings):
# Confirmation link # Confirmation link
CONFIRM_SECRET: str = secrets.token_hex(32) # override in production .env CONFIRM_SECRET: str = secrets.token_hex(32) # override in production .env
CONFIRM_BASE_URL: str = "https://waitlist.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,6 +1,6 @@
import asyncio import asyncio
import datetime
import logging import logging
from urllib.parse import urlencode
from fastapi import APIRouter, Depends, Request from fastapi import APIRouter, Depends, Request
from fastapi.responses import HTMLResponse from fastapi.responses import HTMLResponse
@@ -32,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(
@@ -47,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()
@@ -64,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)
@@ -91,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])
@@ -100,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(
if success: token: str,
title = "You&#8217;re confirmed!" db: AsyncSession = Depends(get_db),
message = "Your email has been verified. We&#8217;ll notify you when adiuvAI is ready." ) -> 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:
title = t["removed_title"]
message = t["removed_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_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_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_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">
@@ -200,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 = "You're on the list!" 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

@@ -57,7 +57,7 @@ async def test_join_waitlist_success(client):
assert resp.status_code == 200 assert resp.status_code == 200
data = resp.json() data = resp.json()
assert data["ok"] is True assert data["ok"] is True
assert "list" in data["message"].lower() assert "inbox" in data["message"].lower()
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -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