"""Email domains domain — business logic. Two responsibilities kept in one module because they share the same workspace scope but are otherwise independent: - ``EmailDomainService`` — SES identity/Configuration Set lifecycle (boto3 SESv2), same client pattern as ``SesProvider`` in ``src/integrations/email_provider.py``. - ``EmailQuotaService`` — monthly/daily send-quota gating, called only from ``src/domain/cloud/email_delivery.py``'s ``platform_domain`` branch. Pure DB logic, no AWS calls. """ from __future__ import annotations import asyncio import re from dataclasses import dataclass from datetime import datetime, timezone from typing import Any, Optional from uuid import UUID import structlog from src.config import Settings, get_settings from src.core.exceptions import NotFoundError, ValidationError from src.domain.billing.access_policy import workspace_has_active_billing from src.domain.billing.plan_quota_service import PlanQuotaService from src.domain.billing.repository import SubscriptionRepository from src.domain.email_domains.models import WorkspaceEmailDomain, WorkspaceEmailQuota from src.domain.email_domains.repository import EmailDomainRepository, EmailQuotaRepository from src.domain.email_domains.schemas import DnsInstructions, DnsRecord logger = structlog.get_logger() _DOMAIN_RE = re.compile( r'^[a-z0-9]([a-z0-9\-]{0,61}[a-z0-9])?' r'(\.[a-z0-9]([a-z0-9\-]{0,61}[a-z0-9])?)+$' ) class EmailDomainError(RuntimeError): pass def _normalize_domain(raw: str) -> str: d = raw.strip().lower() for prefix in ("https://", "http://"): if d.startswith(prefix): d = d[len(prefix):] return d.rstrip("/").rstrip(".").strip() def build_dns_instructions( domain: str, dkim_tokens: list[str], mail_from_domain: str | None, region: str ) -> DnsInstructions: """Build the DKIM (+ optional MAIL FROM) records the user must add at their registrar.""" records: list[DnsRecord] = [] for i, token in enumerate(dkim_tokens, start=1): records.append(DnsRecord( step=i, purpose="dkim", type="CNAME", host=f"{token}._domainkey.{domain}", value=f"{token}.dkim.amazonses.com", ttl=300, note="Proves you control this domain and lets P2HS sign mail on its behalf.", )) if mail_from_domain: next_step = len(records) + 1 records.append(DnsRecord( step=next_step, purpose="mail_from_mx", type="MX", host=mail_from_domain, value=f"feedback-smtp.{region}.amazonses.com", priority=10, ttl=300, note="Routes bounces/complaints for this custom MAIL FROM domain back to SES.", )) records.append(DnsRecord( step=next_step + 1, purpose="mail_from_spf", type="TXT", host=mail_from_domain, value="v=spf1 include:amazonses.com ~all", ttl=300, note="Authorizes SES to send on behalf of this domain (SPF).", )) return DnsInstructions(records=records) @dataclass(frozen=True) class QuotaDecision: allowed: bool reason_code: Optional[str] = None message: Optional[str] = None class EmailDomainService: def __init__( self, domain_repo: EmailDomainRepository, settings: Optional[Settings] = None, *, ses_client_factory: Optional[Any] = None, ): self._repo = domain_repo self._settings = settings or get_settings() self._ses_client_factory = ses_client_factory def _client(self): if self._ses_client_factory is not None: return self._ses_client_factory() import boto3 kwargs: dict[str, str] = {"region_name": self._settings.cloud_ses_region or "us-east-1"} if self._settings.cloud_ses_access_key_id and self._settings.cloud_ses_secret_access_key: kwargs["aws_access_key_id"] = self._settings.cloud_ses_access_key_id kwargs["aws_secret_access_key"] = self._settings.cloud_ses_secret_access_key return boto3.client("sesv2", **kwargs) # ── Sync boto3 calls, always run via asyncio.to_thread ────────────────────────────── def _create_identity_sync(self, domain: str) -> dict: return self._client().create_email_identity( EmailIdentity=domain, DkimSigningAttributes={"NextSigningKeyLength": "RSA_2048_BIT"}, ) def _get_identity_sync(self, domain: str) -> dict: return self._client().get_email_identity(EmailIdentity=domain) def _delete_identity_sync(self, domain: str) -> None: self._client().delete_email_identity(EmailIdentity=domain) def _create_configuration_set_sync(self, name: str) -> None: self._client().create_configuration_set(ConfigurationSetName=name) def _delete_configuration_set_sync(self, name: str) -> None: self._client().delete_configuration_set(ConfigurationSetName=name) def _put_mail_from_sync(self, domain: str, mail_from_domain: str) -> None: self._client().put_email_identity_mail_from_attributes( EmailIdentity=domain, MailFromDomain=mail_from_domain, BehaviorOnMxFailure="USE_DEFAULT_VALUE", ) # ── Domain lifecycle ───────────────────────────────────────────────────────────────── async def add_domain( self, workspace_id: UUID, domain: str, created_by: Optional[UUID] = None ) -> WorkspaceEmailDomain: domain = _normalize_domain(domain) if not _DOMAIN_RE.match(domain): raise ValidationError("Invalid domain name. Use a valid hostname like mail.example.com.") existing_for_domain = await self._repo.get_by_domain(domain) if existing_for_domain: raise ValidationError( "This domain is already claimed by another workspace." if str(existing_for_domain.workspace_id) != str(workspace_id) else "This domain is already added to this workspace." ) existing_for_workspace = await self._repo.get_by_workspace(workspace_id) if existing_for_workspace: raise ValidationError( "This workspace already has a managed email domain " f"({existing_for_workspace.domain}). Remove it before adding a new one." ) configuration_set_name = f"ws-{workspace_id}" try: await asyncio.to_thread(self._create_configuration_set_sync, configuration_set_name) except Exception as exc: if "AlreadyExists" not in type(exc).__name__ and "already exists" not in str(exc).lower(): raise EmailDomainError(f"Could not create SES configuration set: {exc}") from exc try: result = await asyncio.to_thread(self._create_identity_sync, domain) except Exception as exc: raise EmailDomainError(f"Could not register domain with SES: {exc}") from exc dkim_tokens = list((result.get("DkimAttributes") or {}).get("Tokens") or []) d = await self._repo.create( workspace_id, domain, dkim_tokens=dkim_tokens, configuration_set_name=configuration_set_name, created_by=created_by, ) logger.info("email_domain_added", workspace_id=str(workspace_id), domain=domain) return d def dns_instructions_for(self, d: WorkspaceEmailDomain) -> DnsInstructions: return build_dns_instructions( d.domain, list(d.dkim_tokens or []), d.mail_from_domain, self._settings.cloud_ses_region or "us-east-1" ) async def check_verification(self, domain_id: UUID) -> WorkspaceEmailDomain: """Poll SES for DKIM verification. SES is the source of truth — no self DNS lookup.""" d = await self._repo.get_by_id(domain_id) if not d: raise NotFoundError("Email domain not found") if d.status == "verified": return d try: result = await asyncio.to_thread(self._get_identity_sync, d.domain) except Exception as exc: logger.warning("email_domain_ses_status_check_failed", domain=d.domain, error=str(exc)) return d dkim_status = (result.get("DkimAttributes") or {}).get("Status") verified = bool(result.get("VerifiedForSendingStatus")) or dkim_status == "SUCCESS" if verified: d = await self._repo.mark_verified(domain_id) logger.info("email_domain_verified", domain=d.domain, workspace_id=str(d.workspace_id)) elif dkim_status == "FAILED": d = await self._repo.mark_failed(domain_id, "SES reported DKIM verification failed.") else: d.status = "verifying" if d.status == "pending_dns" else d.status return d async def set_mail_from_domain(self, domain_id: UUID, mail_from_domain: str) -> WorkspaceEmailDomain: d = await self._repo.get_by_id(domain_id) if not d: raise NotFoundError("Email domain not found") mail_from_domain = _normalize_domain(mail_from_domain) if not mail_from_domain.endswith(f".{d.domain}"): raise ValidationError(f"MAIL FROM domain must be a subdomain of {d.domain}.") try: await asyncio.to_thread(self._put_mail_from_sync, d.domain, mail_from_domain) except Exception as exc: raise EmailDomainError(f"Could not set MAIL FROM domain: {exc}") from exc return await self._repo.set_mail_from(domain_id, mail_from_domain=mail_from_domain, status="pending") async def delete_domain(self, domain_id: UUID) -> None: d = await self._repo.get_by_id(domain_id) if not d: return try: await asyncio.to_thread(self._delete_identity_sync, d.domain) except Exception as exc: logger.warning("email_domain_ses_delete_failed", domain=d.domain, error=str(exc)) try: await asyncio.to_thread(self._delete_configuration_set_sync, d.ses_configuration_set_name) except Exception as exc: logger.warning("email_domain_ses_configset_delete_failed", domain=d.domain, error=str(exc)) await self._repo.delete(domain_id) logger.info("email_domain_deleted", domain=d.domain, workspace_id=str(d.workspace_id)) class EmailQuotaService: """Monthly/daily quota gate for the ``platform_domain`` email delivery scope. Called once per authorized ``email.send``/``reminder.send`` job — see ``src/domain/cloud/email_delivery.py::resolve_email_delivery``. Never called for the ``connector`` (BYOK) or ``internal`` (shared default sender) delivery scopes. """ def __init__( self, quota_repo: EmailQuotaRepository, plan_quota: PlanQuotaService, subscription_repo: SubscriptionRepository, settings: Optional[Settings] = None, ): self._repo = quota_repo self._plan_quota = plan_quota self._subscription_repo = subscription_repo self._settings = settings or get_settings() @staticmethod def _period_bounds(now: datetime) -> tuple[datetime, datetime]: start = now.replace(day=1, hour=0, minute=0, second=0, microsecond=0) if start.month == 12: end = start.replace(year=start.year + 1, month=1) else: end = start.replace(month=start.month + 1) return start, end async def _load_or_init(self, workspace_id: UUID) -> WorkspaceEmailQuota: q = await self._repo.get_for_update(workspace_id) now = datetime.now(timezone.utc) today = now.date() if q is None: period_start, period_end = self._period_bounds(now) included = await self._plan_quota.email_included_monthly(workspace_id) daily_cap = await self._plan_quota.email_daily_cap(workspace_id) return await self._repo.create( workspace_id, period_start=period_start, period_end=period_end, included_monthly=included, daily_cap=daily_cap, day_bucket_date=today, ) if today != q.day_bucket_date: q.consumed_today = 0 q.day_bucket_date = today if now >= q.period_end: period_start, period_end = self._period_bounds(now) q.period_start = period_start q.period_end = period_end q.consumed_monthly = 0 # Refresh plan-derived limits on period rollover in case the plan changed; # an active override always wins regardless. q.included_monthly = await self._plan_quota.email_included_monthly(workspace_id) q.daily_cap = await self._plan_quota.email_daily_cap(workspace_id) return q async def _has_billable_subscription(self, workspace_id: UUID) -> bool: sub = await self._subscription_repo.get_by_workspace(workspace_id) return bool(sub and workspace_has_active_billing(getattr(sub, "status", None))) async def get_status(self, workspace_id: UUID) -> WorkspaceEmailQuota: """Load/roll the quota row forward for display — never increments. Safe to call as often as a dashboard usage meter wants.""" return await self._load_or_init(workspace_id) async def check_and_reserve(self, workspace_id: UUID) -> QuotaDecision: q = await self._load_or_init(workspace_id) effective_daily_cap = q.override_daily_cap if q.override_daily_cap is not None else q.daily_cap effective_included = ( q.override_included_monthly if q.override_included_monthly is not None else q.included_monthly ) if effective_daily_cap and q.consumed_today >= effective_daily_cap: return QuotaDecision( allowed=False, reason_code="email_daily_cap_exceeded", message=f"Daily sending limit of {effective_daily_cap} reached for this domain.", ) if q.consumed_monthly >= effective_included: if not await self._has_billable_subscription(workspace_id): return QuotaDecision( allowed=False, reason_code="email_quota_exceeded_no_payment_method", message=( f"Included monthly email quota ({effective_included}) is used up. " "Add a payment method or upgrade your plan to keep sending." ), ) q.pending_overage_units += 1 q.consumed_monthly += 1 q.consumed_today += 1 return QuotaDecision(allowed=True) async def set_override( self, workspace_id: UUID, *, included_monthly: Optional[int], daily_cap: Optional[int], rate_per_1000_minor: Optional[int], reason: str, approved_by: UUID, ) -> WorkspaceEmailQuota: return await self._repo.set_override( workspace_id, included_monthly=included_monthly, daily_cap=daily_cap, rate_per_1000_minor=rate_per_1000_minor, reason=reason, approved_by=approved_by, )