"""Templated email dispatch engine (INDL-54). Single entry point — `email_dispatch_service.send(trigger_key, ...)` — that every transactional-email trigger site calls. It: 1. Looks up the tenant's `email_templates` row for the trigger (tenant-scoped). 2. Skips the send (no error) if the admin toggled that template `inactive`. 3. Falls back to a hardcoded default (and self-heals by inserting the missing row) when no row exists at all, so a lookup gap never drops an email. 4. Renders `{{merge_field}}` placeholders via plain string substitution (no Jinja2 / eval — reuses INDL-32's safe approach). Missing fields become "—" and are logged, never left as raw `{{field}}` text. 5. Delegates the physical send to the SMTP-based `_dispatch()` in core.email — SMTP is the one standardized transport (no SES / SendGrid). 6. Writes exactly one `audit_logs` row per attempt (sent / skipped-inactive / default-fallback / failed) with a masked recipient (SEC PII rule). Fail-open (AC-12): `send()` never raises to its caller — a bad template or a dead SMTP relay must not roll back or block the triggering business transaction. All failures are caught, logged, and audited. """ from __future__ import annotations import html import logging import re from dataclasses import dataclass from typing import Literal, Optional from uuid import UUID from pydantic import BaseModel from sqlalchemy.ext.asyncio import AsyncSession from src.apps.settings.services.email_template_service import ( SYSTEM_TEMPLATES, EmailTemplateService, ) from src.apps.site_admin.models.audit_log import AuditLog from src.core.constants import EmailTriggerKey from src.core.email import _dispatch, sanitize_header_value logger = logging.getLogger(__name__) # Matches {{ field_name }} with optional surrounding whitespace; field names are # lowercase-alpha-led identifiers (same shape the INDL-32 schema validator allows). _MERGE_FIELD_RE = re.compile(r"\{\{\s*([a-zA-Z][a-zA-Z0-9_]*)\s*\}\}") TemplateSource = Literal["tenant_custom", "tenant_system", "hardcoded_default"] # ── Hardcoded default subject/body per trigger (Lookup precedence case 3) ────── # Single source of truth is SYSTEM_TEMPLATES (also what the seed migration and # per-tenant provisioning use), keyed here by the trigger enum. DEFAULT_TEMPLATES: dict[EmailTriggerKey, dict] = { EmailTriggerKey(spec["template_key"]): spec for spec in SYSTEM_TEMPLATES } @dataclass(frozen=True) class TriggerMeta: """Static catalog metadata for a trigger (surfaced by GET /triggers).""" label: str recipient_type: str is_implemented: bool # Human-readable catalog. `is_implemented=False` marks triggers whose template is # seeded but for which no domain event fires yet (INDL-54 Out of Scope). TRIGGER_META: dict[EmailTriggerKey, TriggerMeta] = { EmailTriggerKey.NEW_USER_WELCOME: TriggerMeta( "New tenant signup or admin provisioning", "New tenant admin", True), EmailTriggerKey.STAFF_INVITATION: TriggerMeta( "Staff member invited to a tenant", "Invited staff email", True), EmailTriggerKey.STAFF_INVITATION_RESEND: TriggerMeta( "Staff invitation resent", "Invited staff email", True), EmailTriggerKey.CONTACT_ENQUIRY_REPLY: TriggerMeta( "Site admin replies to a contact-form enquiry", "Enquiry submitter", True), EmailTriggerKey.LEAD_QUALIFIED: TriggerMeta( "Enquiry marked qualified → signup link", "Enquiry submitter", True), EmailTriggerKey.PROPOSAL_SENT: TriggerMeta( "Sales proposal / quote sent", "Prospect", True), EmailTriggerKey.PROPOSAL_RESENT: TriggerMeta( "Proposal re-sent after renewal", "Prospect", True), EmailTriggerKey.CONTRACT_SIGNED: TriggerMeta( "Contract issued → signed contract + payment link", "Purchaser", True), EmailTriggerKey.CONTRACT_PDF_CONFIRMATION: TriggerMeta( "Signed contract PDF generated → confirmation", "Purchaser", True), EmailTriggerKey.INVOICE_PAYMENT_RECEIVED: TriggerMeta( "Payment received → receipt", "Purchaser", True), EmailTriggerKey.INVOICE_OVERDUE_REMINDER: TriggerMeta( "Invoice overdue (7 / 14 / 30 days)", "Purchaser", True), EmailTriggerKey.SERVICE_SCHEDULED_FAMILY: TriggerMeta( "Service scheduled → family confirmation", "Family contact", True), EmailTriggerKey.SERVICE_REMINDER_24H: TriggerMeta( "24h before a scheduled service → reminder", "Family contact", True), EmailTriggerKey.FIELD_CREW_BRIEFING: TriggerMeta( "Service scheduled → briefing to field staff", "Assigned staff", False), EmailTriggerKey.TRIBUTE_MODERATED: TriggerMeta( "Tribute approved / rejected", "Tribute submitter", False), EmailTriggerKey.DEED_TRANSFERRED: TriggerMeta( "Plot deed transferred", "New + previous holder", False), EmailTriggerKey.NEWS_NOTIFICATION: TriggerMeta( "News post published", "Subscribers", True), } # Belt-and-braces: every trigger has a default template and catalog entry. assert set(DEFAULT_TEMPLATES) == set(EmailTriggerKey), "DEFAULT_TEMPLATES missing a trigger" assert set(TRIGGER_META) == set(EmailTriggerKey), "TRIGGER_META missing a trigger" class EmailDispatchResult(BaseModel): sent: bool trigger_key: str template_source: TemplateSource skipped_reason: Optional[Literal["template_inactive"]] = None @dataclass class _ResolvedTemplate: subject: str body: str status: str source: TemplateSource template_id: Optional[UUID] def cemetery_context(account) -> dict: """Standard global merge fields (cemetery_name/email/phone + current_date). Every trigger site should spread this into its context so the four global merge fields present in every template render with real values. `account` is a tenants Account row (or None for platform-level triggers). """ from datetime import datetime, timezone return { "cemetery_name": (getattr(account, "organization_name", None) or "our cemetery"), "cemetery_email": getattr(account, "contact_email", None) or "", "cemetery_phone": getattr(account, "contact_phone", None) or "", "current_date": datetime.now(timezone.utc).strftime("%B %d, %Y"), } def _mask_recipient(to: Optional[str]) -> str: """Mask a recipient for logs/audit — never store full family/staff PII.""" if not to or "@" not in to: return "***" local, _, domain = to.partition("@") prefix = local[0] if local else "" return f"{prefix}***@{domain}" class EmailDispatchService: """Trigger → template → render → SMTP send, with audit + fail-open.""" def _render(self, text: str, context: dict) -> tuple[str, set[str]]: """Substitute `{{field}}` from context. Returns (rendered, missing_fields). Missing/blank fields render as "—" (never raw `{{field}}` in a sent email — AC-05) and are collected for logging. """ missing: set[str] = set() def _replace(match: re.Match) -> str: field = match.group(1) value = context.get(field) if value is None or value == "": missing.add(field) return "—" return str(value) return _MERGE_FIELD_RE.sub(_replace, text or ""), missing async def _get_or_seed_template( self, db: AsyncSession, tenant_id: UUID, trigger_key: EmailTriggerKey ) -> _ResolvedTemplate: row = await EmailTemplateService.get_by_trigger_key( db, tenant_id, trigger_key.value ) if row is not None: return _ResolvedTemplate( subject=row.subject, body=row.body, status=row.status, source="tenant_system" if row.is_system else "tenant_custom", template_id=row.id, ) # Case 3: no row at all — send the hardcoded default and self-heal by # inserting the missing system row so future lookups resolve normally. default = DEFAULT_TEMPLATES[trigger_key] seeded = await EmailTemplateService.seed_missing_system_template( db, tenant_id, default ) logger.warning( "[email_dispatch] no template row for tenant=%s trigger=%s — " "seeded default and sent hardcoded fallback", tenant_id, trigger_key.value, ) return _ResolvedTemplate( subject=default["subject"], body=default["body"], status="active", source="hardcoded_default", template_id=seeded.id if seeded else None, ) async def _audit( self, db: AsyncSession, tenant_id: UUID, trigger_key: EmailTriggerKey, to: Optional[str], *, action: str, template_source: TemplateSource, template_id: Optional[UUID], ) -> None: """Write one audit row. Best-effort — never raises to the caller. The insert runs inside a SAVEPOINT (begin_nested) so that, even if the flush fails, only the savepoint is rolled back — the caller's outer transaction stays usable. Without this, a failed flush on the shared HTTP request session would poison the request transaction and make the end-of-request commit raise, aborting the triggering business operation (violates AC-12 fail-open for synchronous callers). """ try: async with db.begin_nested(): db.add( AuditLog( tenant_id=tenant_id, entity_type="email_dispatch", entity_id=template_id, action=action, user_id=None, new_value={ "trigger_key": trigger_key.value, "template_source": template_source, "recipient": _mask_recipient(to), "outcome": action, }, ) ) except Exception as exc: # noqa: BLE001 logger.error("[email_dispatch] audit write failed for %s: %s", trigger_key.value, exc) async def send( self, db: AsyncSession, *, trigger_key: EmailTriggerKey, tenant_id: UUID | str | None, to: str, context: dict, cc: Optional[str] = None, ) -> EmailDispatchResult: """Resolve, render and send the templated email for `trigger_key`. `tenant_id` may be None for platform-level triggers (e.g. a site-admin reply to a public enquiry, which has no owning tenant) — those always render from DEFAULT_TEMPLATES and are audited with a null tenant. Fail-open: any failure is caught, logged and audited; this method never raises, so the triggering business transaction always completes (AC-12). """ key = trigger_key.value if not to: logger.warning("[email_dispatch] no recipient for trigger=%s tenant=%s — skipping", key, tenant_id) return EmailDispatchResult( sent=False, trigger_key=key, template_source="hardcoded_default" ) try: if isinstance(tenant_id, str): tenant_id = UUID(tenant_id) # inside try: a bad value must not raise to the caller if tenant_id is None: default = DEFAULT_TEMPLATES[trigger_key] resolved = _ResolvedTemplate( subject=default["subject"], body=default["body"], status="active", source="hardcoded_default", template_id=None, ) else: resolved = await self._get_or_seed_template(db, tenant_id, trigger_key) except Exception as exc: # noqa: BLE001 logger.error("[email_dispatch] resolve failed trigger=%s tenant=%s: %s", key, tenant_id, exc) await self._audit(db, tenant_id, trigger_key, to, action="email.send_failed", template_source="hardcoded_default", template_id=None) return EmailDispatchResult( sent=False, trigger_key=key, template_source="hardcoded_default" ) # Case 2: admin deliberately turned this notification off — do not send, # do not fall back to a default. This is correct behavior, not a bug. if resolved.status == "inactive": await self._audit(db, tenant_id, trigger_key, to, action="email.skipped_inactive", template_source=resolved.source, template_id=resolved.template_id) return EmailDispatchResult( sent=False, trigger_key=key, template_source=resolved.source, skipped_reason="template_inactive", ) subject, missing_subj = self._render(resolved.subject, context) body, missing_body = self._render(resolved.body, context) subject = sanitize_header_value(subject, "(no subject)") missing = missing_subj | missing_body if missing: logger.warning( "[email_dispatch] missing merge fields for trigger=%s (to=%s): %s", key, _mask_recipient(to), sorted(missing), ) action = ("email.sent_default_fallback" if resolved.source == "hardcoded_default" else "email.sent") try: html_body = html.escape(body).replace("\n", "
\n") await _dispatch(to=to, subject=subject, body_text=body, body_html=html_body, cc=cc) except Exception as exc: # noqa: BLE001 logger.error("[email_dispatch] send failed trigger=%s to=%s: %s", key, _mask_recipient(to), exc) await self._audit(db, tenant_id, trigger_key, to, action="email.send_failed", template_source=resolved.source, template_id=resolved.template_id) return EmailDispatchResult( sent=False, trigger_key=key, template_source=resolved.source ) await self._audit(db, tenant_id, trigger_key, to, action=action, template_source=resolved.source, template_id=resolved.template_id) return EmailDispatchResult( sent=True, trigger_key=key, template_source=resolved.source ) async def list_trigger_catalog( self, db: AsyncSession, tenant_id: UUID ) -> list[dict]: """Return the canonical catalog + this tenant's current mapping status. Powers GET /settings/email-templates/triggers and the admin editor's "Triggered by" line. """ rows = await EmailTemplateService.list_for_tenant(db, tenant_id) by_key = {r.template_key: r for r in rows if r.template_key} catalog: list[dict] = [] for trigger in EmailTriggerKey: meta = TRIGGER_META[trigger] row = by_key.get(trigger.value) catalog.append({ "trigger_key": trigger.value, "label": meta.label, "recipient_type": meta.recipient_type, "is_implemented": meta.is_implemented, "template_id": str(row.id) if row else None, "status": row.status if row else None, }) return catalog # Module-level singleton used by every trigger site. email_dispatch_service = EmailDispatchService()