diff --git a/.env.example b/.env.example
index 6f72742..f280375 100644
--- a/.env.example
+++ b/.env.example
@@ -9,3 +9,14 @@ RATE_LIMIT_PER_MINUTE=5
# Set to "production" in prod to enforce strict origin checks
ENVIRONMENT=development
+
+# Brevo (email) — leave BREVO_API_KEY empty to disable email features
+BREVO_API_KEY=
+BREVO_SENDER_EMAIL=noreply@adiuvai.com
+BREVO_SENDER_NAME=adiuvAI
+BREVO_LIST_ID=0
+
+# Confirmation link
+CONFIRM_SECRET=replace-with-a-long-random-string
+CONFIRM_BASE_URL=https://waitlist.adiuvai.com
+CONFIRM_TOKEN_EXPIRY_HOURS=48
diff --git a/app/brevo.py b/app/brevo.py
new file mode 100644
index 0000000..aa7a57c
--- /dev/null
+++ b/app/brevo.py
@@ -0,0 +1,224 @@
+"""
+Brevo (ex-Sendinblue) integration.
+
+- Send transactional confirmation emails
+- Sync confirmed contacts to a Brevo list
+"""
+
+import logging
+
+import httpx
+
+from app.config import settings
+
+logger = logging.getLogger(__name__)
+
+BREVO_API = "https://api.brevo.com/v3"
+
+
+def _headers() -> dict[str, str]:
+ return {
+ "api-key": settings.BREVO_API_KEY,
+ "Content-Type": "application/json",
+ "Accept": "application/json",
+ }
+
+
+async def send_confirmation_email(email: str, confirm_url: str) -> bool:
+ """Send a double opt-in confirmation email. Returns True on success."""
+ if not settings.brevo_configured:
+ logger.warning("Brevo not configured — skipping confirmation email for %s***", email[:3])
+ return False
+
+ payload = {
+ "sender": {
+ "name": settings.BREVO_SENDER_NAME,
+ "email": settings.BREVO_SENDER_EMAIL,
+ },
+ "to": [{"email": email}],
+ "subject": "Confirm your spot on the adiuvAI waitlist",
+ "htmlContent": _confirmation_html(confirm_url),
+ }
+
+ try:
+ async with httpx.AsyncClient(timeout=10) as client:
+ resp = await client.post(f"{BREVO_API}/smtp/email", headers=_headers(), json=payload)
+ resp.raise_for_status()
+ logger.info("Confirmation email sent to %s***", email[:3])
+ return True
+ except httpx.HTTPError:
+ logger.exception("Failed to send confirmation email to %s***", email[:3])
+ return False
+
+
+async def add_contact_to_list(email: str) -> bool:
+ """Add a confirmed contact to the Brevo waitlist list. Returns True on success."""
+ if not settings.brevo_configured:
+ logger.warning("Brevo not configured — skipping contact sync for %s***", email[:3])
+ return False
+
+ if settings.BREVO_LIST_ID == 0:
+ logger.warning("BREVO_LIST_ID not set — skipping contact sync")
+ return False
+
+ payload = {
+ "email": email,
+ "listIds": [settings.BREVO_LIST_ID],
+ "updateEnabled": True,
+ }
+
+ try:
+ async with httpx.AsyncClient(timeout=10) as client:
+ resp = await client.post(f"{BREVO_API}/contacts", headers=_headers(), json=payload)
+ resp.raise_for_status()
+ logger.info("Contact synced to Brevo list %d: %s***", settings.BREVO_LIST_ID, email[:3])
+ return True
+ except httpx.HTTPError:
+ logger.exception("Failed to sync contact to Brevo: %s***", email[:3])
+ return False
+
+
+def _confirmation_html(confirm_url: str) -> str:
+ """Email template aligned with the adiuvAI landing page brand."""
+ return f"""\
+
+
+
+
+
+
+
+
+ Confirm your spot — adiuvAI
+
+
+
+
+
+
+ Confirm your email to secure your early access spot on the adiuvAI waitlist.
+
+
+
+
+
+
+
+
+
+
+ | |
+
+
+
+
+
+
+ |
+
+
+
+
+ |
+
+ adiuvAI
+ |
+
+
+ |
+
+
+
+
+
+
+
+
+
+ Confirm your spot on
the waitlist
+
+
+
+
+ Thanks for signing up! Please confirm your email address
+ so we can keep you in the loop when adiuvAI launches.
+
+
+
+
+
+
+
+ This link expires in {settings.CONFIRM_TOKEN_EXPIRY_HOURS} hours.
+
+
+ |
+
+
+ |
+
+ |
+
+
+ |
+
+ If you didn't sign up, simply ignore this email.
+
+
+ adiuvai.com
+
+ |
+
+
+
+
+ |
+
+
+
+"""
diff --git a/app/config.py b/app/config.py
index 8e97382..9ae62d7 100644
--- a/app/config.py
+++ b/app/config.py
@@ -1,3 +1,5 @@
+import secrets
+
from pydantic_settings import BaseSettings
@@ -7,6 +9,17 @@ class Settings(BaseSettings):
RATE_LIMIT_PER_MINUTE: int = 5
ENVIRONMENT: str = "development"
+ # Brevo (email)
+ BREVO_API_KEY: str = ""
+ BREVO_SENDER_EMAIL: str = "noreply@adiuvai.com"
+ BREVO_SENDER_NAME: str = "adiuvAI"
+ BREVO_LIST_ID: int = 0 # Brevo contact list ID for waitlist subscribers
+
+ # Confirmation link
+ CONFIRM_SECRET: str = secrets.token_hex(32) # override in production .env
+ CONFIRM_BASE_URL: str = "https://waitlist.adiuvai.com"
+ CONFIRM_TOKEN_EXPIRY_HOURS: int = 48
+
model_config = {"env_file": ".env", "env_file_encoding": "utf-8"}
@property
@@ -17,5 +30,9 @@ class Settings(BaseSettings):
def sync_database_url(self) -> str:
return self.DATABASE_URL.replace("+asyncpg", "+psycopg2")
+ @property
+ def brevo_configured(self) -> bool:
+ return bool(self.BREVO_API_KEY)
+
settings = Settings()
diff --git a/app/routes.py b/app/routes.py
index 279f15e..4b2e5e9 100644
--- a/app/routes.py
+++ b/app/routes.py
@@ -1,13 +1,19 @@
+import asyncio
import logging
+from urllib.parse import urlencode
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()
@@ -25,6 +31,7 @@ async def join_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).
"""
# Honeypot — bots fill hidden fields; silently "succeed"
if body.website:
@@ -35,14 +42,101 @@ async def join_waitlist(
# Check for existing entry — idempotent
existing = await db.execute(
- select(WaitlistEntry.id).where(WaitlistEntry.email == email)
+ select(WaitlistEntry).where(WaitlistEntry.email == email)
)
- if existing.scalar_one_or_none() is not None:
+ entry = existing.scalar_one_or_none()
+
+ if entry is not None:
+ # Re-send confirmation if not yet confirmed
+ if not entry.confirmed and settings.brevo_configured:
+ token = generate_token(email)
+ confirm_url = f"{settings.CONFIRM_BASE_URL}/api/v1/waitlist/confirm?token={token}"
+ asyncio.create_task(send_confirmation_email(email, confirm_url))
return WaitlistResponse()
entry = WaitlistEntry(email=email, ip_address=ip, source="website")
db.add(entry)
await db.commit()
- logger.info("New waitlist signup: %s", email[:3] + "***")
+ 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}"
+ asyncio.create_task(send_confirmation_email(email, confirm_url))
+
return WaitlistResponse()
+
+
+@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)
+
+ if not entry.confirmed:
+ entry.confirmed = True
+ 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))
+
+
+def _result_page(*, success: bool) -> str:
+ """Minimal HTML response for confirmation result."""
+ if success:
+ title = "You're confirmed!"
+ message = "Your email has been verified. We'll notify you when adiuvAI is ready."
+ color = "#16a34a"
+ else:
+ title = "Invalid or expired link"
+ message = "This confirmation link is no longer valid. Please sign up again."
+ color = "#dc2626"
+
+ return f"""\
+
+
+
+
+
+ {title} — adiuvAI
+
+
+
+
+"""
diff --git a/app/token.py b/app/token.py
new file mode 100644
index 0000000..1496c9c
--- /dev/null
+++ b/app/token.py
@@ -0,0 +1,60 @@
+"""
+Stateless HMAC-signed confirmation tokens.
+
+Token format: base64url( email : unix_timestamp : hmac_sha256 )
+No database storage needed — the signature proves authenticity.
+"""
+
+import base64
+import hashlib
+import hmac
+import time
+
+from app.config import settings
+
+
+def generate_token(email: str) -> str:
+ """Create a URL-safe confirmation token for the given email."""
+ timestamp = str(int(time.time()))
+ payload = f"{email}:{timestamp}"
+ sig = hmac.new(
+ settings.CONFIRM_SECRET.encode(),
+ payload.encode(),
+ hashlib.sha256,
+ ).hexdigest()
+ return base64.urlsafe_b64encode(f"{payload}:{sig}".encode()).decode()
+
+
+def verify_token(token: str) -> str | None:
+ """
+ Verify a confirmation token. Returns the email if valid, None otherwise.
+ Checks both HMAC signature and expiry.
+ """
+ try:
+ decoded = base64.urlsafe_b64decode(token.encode()).decode()
+ # rsplit from the right: email may contain colons (unlikely but safe)
+ parts = decoded.rsplit(":", 2)
+ if len(parts) != 3:
+ return None
+
+ email, timestamp, sig = parts
+
+ # Verify HMAC
+ expected = hmac.new(
+ settings.CONFIRM_SECRET.encode(),
+ f"{email}:{timestamp}".encode(),
+ hashlib.sha256,
+ ).hexdigest()
+ if not hmac.compare_digest(sig, expected):
+ return None
+
+ # Check expiry
+ age_seconds = time.time() - int(timestamp)
+ if age_seconds > settings.CONFIRM_TOKEN_EXPIRY_HOURS * 3600:
+ return None
+ if age_seconds < 0:
+ return None # future timestamp — tampered
+
+ return email
+ except Exception:
+ return None
diff --git a/requirements.txt b/requirements.txt
index 2e8d622..5d4ecde 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -8,3 +8,4 @@ pydantic>=2.0,<3.0
pydantic-settings>=2.0,<3.0
email-validator>=2.0,<3.0
python-dotenv>=1.0,<2.0
+httpx>=0.27,<1.0
diff --git a/tests/test_waitlist.py b/tests/test_waitlist.py
index 8d9a70f..8d95f6e 100644
--- a/tests/test_waitlist.py
+++ b/tests/test_waitlist.py
@@ -1,12 +1,17 @@
+import time
+from unittest.mock import AsyncMock, patch
+
import pytest
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
+from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from app.main import app
-from app.models import Base
+from app.models import Base, WaitlistEntry
from app.db import get_db
from app.rate_limit import _hits_store
+from app.token import generate_token, verify_token
# Use SQLite for tests (no Postgres dependency)
TEST_DB_URL = "sqlite+aiosqlite:///./test_waitlist.db"
@@ -107,3 +112,115 @@ async def test_rate_limit(client):
# The 6th request should be rate-limited (limit is 5)
assert resp.status_code == 429
assert "Retry-After" in resp.headers
+
+
+# ── Confirmation token tests ─────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_token_roundtrip():
+ """A generated token should verify back to the same email."""
+ email = "token@example.com"
+ token = generate_token(email)
+ assert verify_token(token) == email
+
+
+@pytest.mark.asyncio
+async def test_token_expired():
+ """An expired token should return None."""
+ email = "expired@example.com"
+ with patch("app.token.time") as mock_time:
+ # Generate token "49 hours ago"
+ past = time.time() - 49 * 3600
+ mock_time.time.return_value = past
+ token = generate_token(email)
+
+ # Now verify with real time — should be expired (>48h)
+ assert verify_token(token) is None
+
+
+@pytest.mark.asyncio
+async def test_token_tampered():
+ """A tampered token should return None."""
+ token = generate_token("legit@example.com")
+ # Flip a character in the token
+ tampered = token[:-1] + ("A" if token[-1] != "A" else "B")
+ assert verify_token(tampered) is None
+
+
+# ── Confirm endpoint tests ───────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_confirm_valid_token(client, db_session):
+ """GET /confirm with valid token marks email as confirmed."""
+ # Seed an unconfirmed entry
+ entry = WaitlistEntry(email="confirm@example.com", source="website")
+ db_session.add(entry)
+ await db_session.commit()
+
+ token = generate_token("confirm@example.com")
+ resp = await client.get(f"/api/v1/waitlist/confirm?token={token}")
+ assert resp.status_code == 200
+ assert "confirmed" in resp.text.lower() or "verified" in resp.text.lower()
+
+ # Verify DB state
+ result = await db_session.execute(
+ select(WaitlistEntry).where(WaitlistEntry.email == "confirm@example.com")
+ )
+ assert result.scalar_one().confirmed is True
+
+
+@pytest.mark.asyncio
+async def test_confirm_invalid_token(client):
+ """GET /confirm with invalid token returns 400."""
+ resp = await client.get("/api/v1/waitlist/confirm?token=garbage")
+ assert resp.status_code == 400
+ assert "invalid" in resp.text.lower() or "expired" in resp.text.lower()
+
+
+@pytest.mark.asyncio
+async def test_confirm_idempotent(client, db_session):
+ """Confirming an already confirmed email returns 200 (idempotent)."""
+ entry = WaitlistEntry(email="idem@example.com", source="website", confirmed=True)
+ db_session.add(entry)
+ await db_session.commit()
+
+ token = generate_token("idem@example.com")
+ resp = await client.get(f"/api/v1/waitlist/confirm?token={token}")
+ assert resp.status_code == 200
+
+
+@pytest.mark.asyncio
+async def test_confirm_unknown_email(client):
+ """Token for a non-existent email returns 400."""
+ token = generate_token("unknown@example.com")
+ resp = await client.get(f"/api/v1/waitlist/confirm?token={token}")
+ assert resp.status_code == 400
+
+
+# ── Brevo integration tests (mocked) ────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_signup_triggers_confirmation_email(client, db_session):
+ """When Brevo is configured, signup sends a confirmation email."""
+ with patch("app.routes.settings") as mock_settings, \
+ patch("app.routes.send_confirmation_email", new_callable=AsyncMock) as mock_send:
+ mock_settings.brevo_configured = True
+ mock_settings.CONFIRM_BASE_URL = "http://test"
+
+ resp = await client.post(
+ "/api/v1/waitlist",
+ json={"email": "brevo@example.com"},
+ )
+ assert resp.status_code == 200
+
+ # Wait for fire-and-forget task
+ import asyncio
+ await asyncio.sleep(0.1)
+
+ mock_send.assert_called_once()
+ call_args = mock_send.call_args
+ assert call_args[0][0] == "brevo@example.com"
+ assert "confirm" in call_args[0][1]