import asyncio
import datetime
import logging
from fastapi import APIRouter, Depends, Request
from fastapi.responses import HTMLResponse
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.brevo import add_contact_to_list, send_confirmation_email
from app.config import settings
from app.db import get_db
from app.rate_limit import _get_client_ip
from app.schemas import WaitlistRequest, WaitlistResponse
from app.models import WaitlistEntry
from app.token import generate_token, verify_token
logger = logging.getLogger(__name__)
router = APIRouter()
@router.post("/waitlist", response_model=WaitlistResponse)
async def join_waitlist(
body: WaitlistRequest,
request: Request,
db: AsyncSession = Depends(get_db),
) -> WaitlistResponse:
"""
Add an email to the waitlist.
- Honeypot: if `website` field is non-empty, silently succeed (bot trap).
- Duplicate emails: idempotent — returns success without error.
- Stores the Cloudflare-resolved client IP for analytics (not exposed).
- Sends a confirmation email via Brevo (fire-and-forget).
- Records consent timestamp (GDPR Art. 7).
"""
# Honeypot — bots fill hidden fields; silently "succeed"
if body.website:
return WaitlistResponse.for_lang(body.lang)
email = body.email.lower().strip()
lang = body.lang
ip = _get_client_ip(request)
now = datetime.datetime.now(datetime.timezone.utc)
# Check for existing entry — idempotent
existing = await db.execute(
select(WaitlistEntry).where(WaitlistEntry.email == email)
)
entry = existing.scalar_one_or_none()
if entry is not None:
# Update language preference if changed
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)
confirm_url = f"{settings.CONFIRM_BASE_URL}/api/v1/waitlist/confirm?token={token}"
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.for_lang(lang)
entry = WaitlistEntry(
email=email,
ip_address=ip,
source="website",
language=lang,
consent_given_at=now,
)
db.add(entry)
await db.commit()
logger.info("New waitlist signup: %s***", email[:3])
# Fire-and-forget: send confirmation email via Brevo
if settings.brevo_configured:
token = generate_token(email)
confirm_url = f"{settings.CONFIRM_BASE_URL}/api/v1/waitlist/confirm?token={token}"
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.for_lang(lang)
@router.get("/waitlist/confirm", response_class=HTMLResponse)
async def confirm_email(
token: str,
db: AsyncSession = Depends(get_db),
) -> HTMLResponse:
"""
Double opt-in confirmation endpoint.
Verifies the HMAC token, marks the entry as confirmed,
and syncs the contact to Brevo.
"""
email = verify_token(token)
if email is None:
return HTMLResponse(content=_result_page(success=False), status_code=400)
result = await db.execute(
select(WaitlistEntry).where(WaitlistEntry.email == email)
)
entry = result.scalar_one_or_none()
if entry is None:
return HTMLResponse(content=_result_page(success=False), status_code=400)
lang = entry.language or "en"
if not entry.confirmed:
entry.confirmed = True
# Clear IP now that consent is confirmed — no longer needed
entry.ip_address = None
await db.commit()
logger.info("Email confirmed: %s***", email[:3])
# Sync confirmed contact to Brevo list (fire-and-forget)
if settings.brevo_configured:
asyncio.create_task(add_contact_to_list(email))
return HTMLResponse(content=_result_page(success=True, lang=lang))
@router.get("/waitlist/unsubscribe", response_class=HTMLResponse)
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’re confirmed!",
"confirmed_msg": "Your email has been verified. We’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 e‑mail 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'e‑mails 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 E‑Mail 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 E‑Mails 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 = ''
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 = ''
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 = ''
else:
title = t["invalid_title"]
message = t["invalid_confirm"]
icon_bg = "rgba(220,38,38,0.08)"
icon_border = "rgba(220,38,38,0.18)"
icon_svg = ''
btn_label = t["btn"]
return f"""\
{title} — adiuvAI