Brevo transaktionsmail i SaaS-platform — implementation med Celery

M-Bus Gateway platformen sender 10+ typer transaktionsmails — fra årsafregninger til MID-advarsler. Her er implementeringen med Brevo (tidligere Sendinblue) og best practices for GDPR-compliance.

---

Hvorfor Brevo frem for andre?

Brevo vælges fordi:

• EU-hostet (Frankrig) = GDPR-nemt
• Transaktionsmail + Marketing i ét system
• Gratis unsubscribe-håndtering
• SMTP + REST API

---

Email-oversigt: Alle mail-typer i platformen

# server/src/workers/tasks/email.py
EMAIL_TYPES = {
    # Lejer-vendte:
    "settlement_email": "Årsafregning (PDF vedlagt)",
    "tenant_welcome": "Velkomstmail til ny lejer (GDPR-info)",
    "settlement_reminder": "Påmindelsesmail — afregning ikke åbnet 7 dage",
    "magic_link": "Magic link til lejerportal",

    # Udlejer-vendte:
    "payment_failed": "Betalingsfejl — past_due dunning",
    "renewal_reminder": "Stripe-fornyelse 7 dage frem",
    "trial_ending": "Trial udløber om 3 dage",
    "settlement_deadline": "Afregningsdeadline advarsel",
    "reading_coverage_alert": "Aflæsningsdækning under tærskel",
    "monthly_digest": "Månedlig portefølje-digest",
    "onboarding_drip": "Onboarding drip-serie (dag 1, 3, 7)",
    "mid_expiry": "MID-certifikat udløber om 6 måneder",

    # System:
    "gateway_offline": "Gateway offline > 36 timer",
    "new_tenant_welcome": "Ny kunde (udlejer) oprettet",
}

---

Brevo API-klient

# server/src/email/brevo_client.py
import httpx
from typing import Optional

class BrevoClient:
    BASE_URL = "https://api.brevo.com/v3"

    def __init__(self, api_key: str, sender_email: str, sender_name: str):
        self.api_key = api_key
        self.sender = {"email": sender_email, "name": sender_name}
        self._client = httpx.AsyncClient(
            base_url=self.BASE_URL,
            headers={
                "api-key": api_key,
                "Content-Type": "application/json",
            },
            timeout=15,
        )

    async def send_transactional(
        self,
        to_email: str,
        to_name: str,
        subject: str,
        html_content: str,
        text_content: Optional[str] = None,
        reply_to: Optional[str] = None,
        attachments: Optional[list] = None,
        tags: Optional[list] = None,
    ) -> dict:
        payload = {
            "sender": self.sender,
            "to": [{"email": to_email, "name": to_name}],
            "subject": subject,
            "htmlContent": html_content,
            "tags": tags or [],
        }
        if text_content:
            payload["textContent"] = text_content
        if reply_to:
            payload["replyTo"] = {"email": reply_to}
        if attachments:
            payload["attachment"] = attachments  # [{"name": "...", "content": base64}]

        resp = await self._client.post("/smtp/email", json=payload)
        resp.raise_for_status()
        return resp.json()

---

Celery task: Årsafregning med PDF

# server/src/workers/tasks/email.py
from celery import shared_task
import base64

@shared_task(
    name="send_settlement_email",
    bind=True,
    max_retries=5,
    default_retry_delay=300,
    autoretry_for=(httpx.TimeoutException,),
    retry_backoff=True,
)
def send_settlement_email(
    self,
    settlement_id: str,
    tenant_user_id: str,
):
    import asyncio
    asyncio.run(_send_settlement(settlement_id, tenant_user_id))


async def _send_settlement(settlement_id: str, tenant_user_id: str):
    async with async_session_factory() as session:
        settlement = await session.get(Settlement, settlement_id)
        user = await session.get(User, tenant_user_id)

        # Hent PDF fra Object Storage:
        pdf_bytes = await download_from_storage(settlement.pdf_url)
        pdf_b64 = base64.b64encode(pdf_bytes).decode()

        # Byg HTML via Jinja2:
        html = render_template(
            "email/settlement.html",
            tenant_name=user.full_name,
            period=f"{settlement.period_start.year}/{settlement.period_end.year}",
            balance=settlement.balance_kr,
            magic_link=generate_magic_link(settlement_id, user.id),
            dispute_deadline=settlement.dispute_deadline,
        )

        await brevo.send_transactional(
            to_email=user.email,
            to_name=user.full_name,
            subject=f"Din varmeafregning {settlement.period_start.year}/{settlement.period_end.year}",
            html_content=html,
            attachments=[{
                "name": f"varmeafregning-{settlement_id}.pdf",
                "content": pdf_b64,
            }],
            tags=["settlement", f"tenant-{user.id}"],
        )

        # Registrér sendt:
        settlement.email_sent_at = datetime.utcnow()
        settlement.status = "sent"
        session.add(settlement)
        await session.commit()

---

Sender-domain opsætning

# DNS-records der skal tilføjes for god deliverability:

# SPF:
v=spf1 include:spf.brevo.com ~all

# DKIM:
# Brevo leverer DKIM-key → tilføj som TXT-record:
# mail._domainkey.mbus-gateway.dk → [Brevo DKIM key]

# DMARC:
_dmarc.mbus-gateway.dk → "v=DMARC1; p=quarantine; rua=mailto:dmarc@mbus-gateway.dk"

# Verificér opsætning:
# Brevo dashboard → Senders & IPs → Domains → Verificer

---

GDPR og unsubscribe-håndtering

# Lejer har ret til at afmelde notifikationer:
# User.notification_prefs er en JSON-kolonne:
default_prefs = {
    "settlement_emails": True,    # Kan IKKE afmeldes (juridisk krav)
    "reminder_emails": True,      # Kan afmeldes
    "monthly_digest": True,       # Kan afmeldes
    "gateway_alerts": True,       # Kan afmeldes (udlejere)
    "marketing": False,           # Default fra → skal aktivt tilmeldes
}

# I email-tasks: Tjek altid prefs inden afsendelse:
async def _send_reminder(settlement_id: str, user_id: str):
    user = await get_user(user_id)

    # Tjek opt-in:
    if not user.notification_prefs.get("reminder_emails", True):
        logger.info("Bruger afmeldt påmindelser", user_id=user_id)
        return

    # ... send email

# Unsubscribe link i alle emails:
# <a href="{FRONTEND_URL}/portal/notification-settings?token={jwt}">
#   Afmeld notifikationer</a>

---

Bounce-håndtering via Brevo webhook

# server/src/email/webhook.py
from fastapi import APIRouter, Request

router = APIRouter(prefix="/webhooks/brevo", tags=["email"])

@router.post("/")
async def brevo_webhook(
    request: Request,
    session: AsyncSession = Depends(get_session),
):
    events = await request.json()
    for event in events:
        email = event.get("email")
        event_type = event.get("event")

        if event_type == "hard_bounce":
            # Deaktivér email (permanent fejl):
            await mark_email_invalid(email, session)
            logger.warning("Hard bounce", email=email)

        elif event_type == "unsubscribe":
            # Respektér unsubscribe:
            await disable_marketing_emails(email, session)

---

Konklusion

Brevo er det rette valg til transaktionsmail i en EU-hostet SaaS-platform: GDPR-nativt, sender-domain setup med SPF/DKIM/DMARC, og Celery-integration med retry-logik sikrer at afregninger og alarmer leveres pålideligt. Altid tjek notification_prefs inden afsendelse — og aldrig deaktivér juridisk påkrævede mails (årsafregninger).

Se Celery Beat guide eller WeasyPrint PDF-guide.