from app.email.scaleway import ScalewayEmailClient SUPPORTED_KINDS: frozenset[str] = frozenset( {"verify-email", "verify-email-reminder", "reset-password", "invitation", "invitation-reminder"} ) def _escape_html(value: str) -> str: return ( value.replace("&", "&") .replace("<", "<") .replace(">", ">") .replace('"', """) .replace("'", "'") ) def _layout( *, title: str, intro: str, action_label: str, url: str, details: str, outro: str, ) -> str: return f"""

Dineo

{title}

{intro}

{details}

{action_label}

Or copy and paste this link into your browser:
{url}

{outro}

""" async def send_verify_email( client: ScalewayEmailClient, *, to: str, name: str | None, url: str, ) -> None: safe_name = _escape_html(name.strip()) if name and name.strip() else "there" safe_url = _escape_html(url) subject = "Dineo email verification" html = _layout( title="Confirm your email", intro=f"Hi {safe_name}, thanks for creating your Dineo account.", action_label="Verify email", url=safe_url, details=("Use the button below to verify your email address and finish setting up access."), outro="If you did not request this email, you can ignore it.", ) await client._send(to, subject, html) async def send_reset_password_email( client: ScalewayEmailClient, *, to: str, name: str | None, url: str, ) -> None: safe_name = _escape_html(name.strip()) if name and name.strip() else "there" safe_url = _escape_html(url) subject = "Dineo password reset" html = _layout( title="Reset your password", intro=f"Hi {safe_name}, we received a request to reset your Dineo password.", action_label="Reset password", url=safe_url, details="Use the secure link below to choose a new password for your account.", outro=( "If you did not request a password reset, you can ignore this email and " "your password will stay unchanged." ), ) await client._send(to, subject, html) async def send_invitation_email( client: ScalewayEmailClient, *, to: str, url: str, organization_name: str | None, inviter_email: str | None, ) -> None: safe_url = _escape_html(url) safe_org = ( _escape_html(organization_name.strip()) if organization_name and organization_name.strip() else "your team" ) safe_inviter = ( _escape_html(inviter_email.strip()) if inviter_email and inviter_email.strip() else "a Dineo administrator" ) subject = f"Dineo invitation to join {safe_org}" html = _layout( title="You're invited to Dineo", intro=f"{safe_inviter} invited you to join {safe_org} on Dineo.", action_label="Accept invitation", url=safe_url, details=( "Open the invitation link below to create your account or sign in and " "access the organization." ), outro="If you were not expecting this invitation, you can ignore this email.", ) await client._send(to, subject, html) async def send_verify_email_reminder( client: ScalewayEmailClient, *, to: str, name: str | None, url: str | None, ) -> None: """Nudge sent at +48h by the AuthVerificationWorkflow when the verify link has not been clicked. The ``url`` is a sign-in landing page (BA's own verification token has expired by then); the user signs in there and clicks "resend verification" to receive a fresh BA link.""" safe_name = _escape_html(name.strip()) if name and name.strip() else "there" landing = _escape_html(url) if url else "" subject = "Reminder: confirm your Dineo email" html = _layout( title="Still need to confirm your email", intro=f"Hi {safe_name}, your Dineo account is waiting on email confirmation.", action_label="Sign in to resend", url=landing or "#", details=( "We sent a verification link 48 hours ago but it looks like it has " "not been opened yet. Sign in to request a fresh link — accounts " "without confirmation are removed automatically after seven days." ), outro="If you no longer need the account, you can ignore this email.", ) await client._send(to, subject, html) async def send_invitation_reminder( client: ScalewayEmailClient, *, to: str, url: str, organization_name: str | None, inviter_email: str | None, ) -> None: """Nudge sent halfway through an invitation's expiry window by the InvitationLifecycleWorkflow. Same accept link as the original invite.""" safe_url = _escape_html(url) safe_org = ( _escape_html(organization_name.strip()) if organization_name and organization_name.strip() else "your team" ) safe_inviter = ( _escape_html(inviter_email.strip()) if inviter_email and inviter_email.strip() else "a Dineo administrator" ) subject = f"Reminder: your invitation to {safe_org}" html = _layout( title="Your invitation is still waiting", intro=f"{safe_inviter} invited you to join {safe_org} on Dineo.", action_label="Accept invitation", url=safe_url, details=( "Invitations expire automatically. Open the link below to accept " "before the window closes." ), outro="If you no longer wish to join, you can ignore this email.", ) await client._send(to, subject, html) async def dispatch_auth_email( client: ScalewayEmailClient, *, kind: str, to: str, name: str | None, url: str | None, organization_name: str | None, inviter_email: str | None, ) -> None: if kind == "verify-email": if url is None: raise ValueError("Missing url for verify-email") await send_verify_email(client, to=to, name=name, url=url) return if kind == "verify-email-reminder": await send_verify_email_reminder(client, to=to, name=name, url=url) return if kind == "reset-password": if url is None: raise ValueError("Missing url for reset-password") await send_reset_password_email(client, to=to, name=name, url=url) return if kind == "invitation": if url is None: raise ValueError("Missing url for invitation") await send_invitation_email( client, to=to, url=url, organization_name=organization_name, inviter_email=inviter_email, ) return if kind == "invitation-reminder": if url is None: raise ValueError("Missing url for invitation-reminder") await send_invitation_reminder( client, to=to, url=url, organization_name=organization_name, inviter_email=inviter_email, ) return raise ValueError(f"Unknown kind: {kind}")