"""Email Templates service (INDL-32).
Security posture:
- Every read/write is tenant-scoped (`tenant_id == :tenant_id`); lookups by
id that don't match the tenant return None → router raises a plain 404
(never 403) to avoid tenant enumeration (SEC-07 / API1).
- System templates (`is_system=True`) can never be deleted, and
`is_system` / `template_key` are never mutated by `update()` even if
present on the payload (defense in depth beyond the Pydantic schema —
SEC-08 / API3).
- Every mutating method writes an AuditLog row (SEC-06 / A09).
- `send_test` never uses a template engine — merge fields are substituted
via plain `str.replace` (Risks & Notes / SEC-02 / SEC-09), and returns a
503 if no outbound email transport is configured, rather than silently
"succeeding" by only logging to console.
- `send_test` HTML-escapes the rendered body before passing it as the
`body_html` part of the outbound message (H-1 security fix) — bodies are
plain text, never real HTML, so the raw rendered string must never be
treated as trusted markup even if it passed the schema-level denylist.
"""
from __future__ import annotations
import html
from datetime import datetime, timezone
from typing import Optional
from uuid import UUID
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
from src.apps.auth.models.user import User
from src.apps.settings.models.email_template import EmailTemplate
from src.apps.site_admin.models.audit_log import AuditLog
from src.core.config import settings
from src.core.email import _dispatch
from src.core.exceptions import ConflictError, ForbiddenError, NotFoundError, ServiceUnavailableError
# ── System template seed data (INDL-32 PRD "Seed Data" table) ──────────────────
_GLOBAL_MERGE_FIELDS = ["cemetery_name", "cemetery_email", "cemetery_phone", "current_date"]
# keep in sync with src/database/migrations/versions/0054_email_templates_prd_schema.py:SYSTEM_TEMPLATES
# (migrations are self-contained/frozen-in-time and intentionally keep their own copy of seed data —
# do not import this constant into a migration; update both places if the seed set changes)
SYSTEM_TEMPLATES: list[dict] = [
{
"template_key": "contract_signed",
"name": "Contract signed — confirmation",
"description": "Sent to purchaser when a contract is signed",
"status": "active",
"subject": "Your contract with {{cemetery_name}} is confirmed",
"body": (
"Dear {{purchaser_name}},\n\n"
"Thank you for choosing {{cemetery_name}}. This confirms your "
"{{contract_type}} contract ({{contract_number}}) for {{plot_id}} "
"in {{section}} has been signed.\n\n"
"Contract total: {{contract_total}}\n"
"Balance due: {{balance_due}}\n\n"
"If you have any questions, contact us at {{cemetery_email}} or "
"{{cemetery_phone}}.\n\n"
"Sincerely,\n{{cemetery_name}}"
),
"merge_fields": [
"purchaser_name", "contract_type", "plot_id", "section",
"contract_number", "contract_total", "balance_due",
],
},
{
"template_key": "invoice_payment_received",
"name": "Invoice — payment received",
"description": "Receipt to purchaser",
"status": "active",
"subject": "Payment received — invoice {{invoice_number}}",
"body": (
"Dear {{purchaser_name}},\n\n"
"We have received your payment of {{amount_paid}} on {{payment_date}} "
"for invoice {{invoice_number}}.\n\n"
"Invoice total: {{invoice_total}}\n"
"Remaining balance: {{balance_due}}\n\n"
"Thank you,\n{{cemetery_name}}"
),
"merge_fields": [
"purchaser_name", "invoice_number", "amount_paid", "payment_date",
"invoice_total", "balance_due",
],
},
{
"template_key": "invoice_overdue_reminder",
"name": "Invoice overdue reminder",
"description": "7 / 14 / 30 days overdue",
"status": "active",
"subject": "Reminder: invoice {{invoice_number}} is {{days_overdue}} days overdue",
"body": (
"Dear {{purchaser_name}},\n\n"
"Our records show invoice {{invoice_number}} (total {{invoice_total}}) "
"was due on {{due_date}} and is now {{days_overdue}} days overdue.\n\n"
"Amount due: {{amount_due}}\n\n"
"Please arrange payment at your earliest convenience. Contact us at "
"{{cemetery_email}} or {{cemetery_phone}} with any questions.\n\n"
"Regards,\n{{cemetery_name}}"
),
"merge_fields": [
"purchaser_name", "invoice_number", "invoice_total", "amount_due",
"due_date", "days_overdue",
],
},
{
"template_key": "service_scheduled_family",
"name": "Service scheduled — family notice",
"description": "Interment / memorial confirmation",
"status": "active",
"subject": "Your {{service_type}} at {{cemetery_name}} is confirmed",
"body": (
"Dear {{family_name}},\n\n"
"This confirms the {{service_type}} scheduled for {{service_date}} at "
"{{service_time}}, at {{location}}.\n\n"
"Officiant: {{officiant_name}}\n\n"
"Please contact us at {{cemetery_email}} or {{cemetery_phone}} if you "
"have any questions.\n\n"
"With sympathy,\n{{cemetery_name}}"
),
"merge_fields": [
"family_name", "service_type", "service_date", "service_time",
"location", "officiant_name",
],
},
{
"template_key": "field_crew_briefing",
"name": "Field crew briefing sheet",
"description": "Sent to assigned field staff",
"status": "active",
"subject": "Briefing: {{service_type}} on {{service_date}}",
"body": (
"Team,\n\n"
"You are assigned to a {{service_type}} on {{service_date}} at "
"{{service_time}}.\n\n"
"Location: {{plot_id}}, {{section}}\n"
"Assigned staff: {{assigned_staff}}\n\n"
"— {{cemetery_name}} Operations"
),
"merge_fields": [
"service_type", "service_date", "service_time", "plot_id",
"section", "assigned_staff",
],
},
{
"template_key": "tribute_moderated",
"name": "Tribute approved / rejected",
"description": "To memorial submitter",
"status": "active",
"subject": "Your tribute for {{memorial_name}} has been {{tribute_status}}",
"body": (
"Dear {{submitter_name}},\n\n"
"Your submitted tribute for {{memorial_name}} has been {{tribute_status}}.\n\n"
"{{rejection_reason}}\n\n"
"Thank you for sharing your memories.\n\n"
"{{cemetery_name}}"
),
"merge_fields": [
"submitter_name", "memorial_name", "tribute_status", "rejection_reason",
],
},
{
"template_key": "deed_transferred",
"name": "Deed transferred",
"description": "To new + previous deed holder",
"status": "inactive",
"subject": "Deed transfer confirmation — {{plot_id}}",
"body": (
"This confirms the deed for {{plot_id}} in {{section}} was transferred "
"from {{previous_holder_name}} to {{new_holder_name}} effective "
"{{transfer_date}}.\n\n"
"Questions? Contact us at {{cemetery_email}} or {{cemetery_phone}}.\n\n"
"{{cemetery_name}}"
),
"merge_fields": [
"new_holder_name", "previous_holder_name", "plot_id", "section",
"transfer_date",
],
},
{
"template_key": "new_user_welcome",
"name": "New user welcome",
"description": "Temporary password + sign-in link",
"status": "active",
"subject": "Welcome to {{cemetery_name}} on INDELIS",
"body": (
"Hi {{user_name}},\n\n"
"An account has been created for you on {{cemetery_name}}'s INDELIS "
"workspace.\n\n"
"Temporary password: {{temp_password}}\n"
"Sign in here: {{login_url}}\n\n"
"You will be asked to set a new password on first login.\n\n"
"— {{cemetery_name}}"
),
"merge_fields": ["user_name", "temp_password", "login_url"],
},
# ── INDL-54: 9 additional trigger templates (previously hardcoded, no key) ──
{
"template_key": "staff_invitation",
"name": "Staff invitation",
"description": "Sent to a staff member invited to the tenant",
"status": "active",
"subject": "You've been invited to join {{organization_name}} on INDELIS",
"body": (
"Hi {{invitee_name}},\n\n"
"{{invited_by_name}} has invited you to join {{organization_name}} on "
"INDELIS as a {{role}}.\n\n"
"Accept your invitation and set your password here:\n{{invite_url}}\n\n"
"This link is valid for 7 days and can only be used once.\n\n"
"— {{cemetery_name}}"
),
"merge_fields": [
"invitee_name", "invited_by_name", "organization_name", "role", "invite_url",
],
},
{
"template_key": "staff_invitation_resend",
"name": "Staff invitation — resent",
"description": "Reminder resend of a staff invitation",
"status": "active",
"subject": "Reminder: your invitation to join {{organization_name}} on INDELIS",
"body": (
"Hi {{invitee_name}},\n\n"
"This is a reminder that {{invited_by_name}} invited you to join "
"{{organization_name}} on INDELIS as a {{role}}.\n\n"
"Accept your invitation and set your password here:\n{{invite_url}}\n\n"
"This link is valid for 7 days and can only be used once.\n\n"
"— {{cemetery_name}}"
),
"merge_fields": [
"invitee_name", "invited_by_name", "organization_name", "role", "invite_url",
],
},
{
"template_key": "contact_enquiry_reply",
"name": "Contact enquiry reply",
"description": "Site admin's reply to a public contact-form enquiry",
"status": "active",
"subject": "Re: your enquiry to {{cemetery_name}}",
"body": (
"Hi {{contact_name}},\n\n"
"Thank you for reaching out. Here is a response from our team:\n\n"
"{{reply_message}}\n\n"
"If you have any further questions, please reply to this email.\n\n"
"Warm regards,\n{{admin_name}}\n{{cemetery_name}}"
),
"merge_fields": ["contact_name", "reply_message", "admin_name"],
},
{
"template_key": "lead_qualified",
"name": "Lead qualified — signup invitation",
"description": "Sent when an enquiry is marked qualified, with a signup link",
"status": "active",
"subject": "You're approved — complete your INDELIS setup",
"body": (
"Hi {{contact_name}},\n\n"
"Great news — we've reviewed your enquiry for {{organization}} and "
"you're approved to join INDELIS!\n\n"
"Complete your account setup here:\n{{signup_url}}\n\n"
"Your information has been pre-filled to save you time.\n\n"
"Warm regards,\nThe INDELIS Team"
),
"merge_fields": ["contact_name", "organization", "signup_url"],
},
{
"template_key": "proposal_sent",
"name": "Sales proposal sent",
"description": "Price proposal / quote sent to a prospect",
"status": "active",
"subject": "Your price proposal {{quote_number}} from {{cemetery_name}}",
"body": (
"{{cover_note}}\n\n"
"--- QUOTE {{quote_number}} ---\n"
"{{line_items}}\n\n"
"Subtotal: {{subtotal}}\n"
"Tax: {{tax_amount}}\n"
"Total (CAD): {{total_amount}}\n\n"
"{{expiry_note}}\n\n"
"This is a non-binding price quote. Reply to this email with any questions."
),
"merge_fields": [
"cover_note", "quote_number", "line_items", "subtotal",
"tax_amount", "total_amount", "expiry_note",
],
},
{
"template_key": "proposal_resent",
"name": "Sales proposal — resent",
"description": "Proposal re-sent to a prospect after renewal",
"status": "active",
"subject": "[Resent] Your price proposal {{quote_number}} from {{cemetery_name}}",
"body": (
"{{cover_note}}\n\n"
"--- QUOTE {{quote_number}} ---\n"
"{{line_items}}\n\n"
"Subtotal: {{subtotal}}\n"
"Tax: {{tax_amount}}\n"
"Total (CAD): {{total_amount}}\n\n"
"{{expiry_note}}\n\n"
"This is a non-binding price quote. Reply to this email with any questions."
),
"merge_fields": [
"cover_note", "quote_number", "line_items", "subtotal",
"tax_amount", "total_amount", "expiry_note",
],
},
{
"template_key": "contract_pdf_confirmation",
"name": "Contract PDF confirmation",
"description": "Confirmation sent once a signed contract PDF is generated",
"status": "active",
"subject": "Your contract {{contract_number}} is ready",
"body": (
"Dear {{purchaser_name}},\n\n"
"Your signed Cemetery Plot Purchase Agreement (Contract "
"#{{contract_number}}) has been processed and is ready.\n\n"
"Please retain this confirmation for your records. A copy of your "
"contract PDF is available through your INDELIS portal.\n\n"
"If you have any questions, please contact the cemetery office.\n\n"
"— {{cemetery_name}}"
),
"merge_fields": ["purchaser_name", "contract_number"],
},
{
"template_key": "service_reminder_24h",
"name": "Service reminder (24h)",
"description": "24-hour reminder before a scheduled service",
"status": "active",
"subject": "Reminder: service tomorrow for {{decedent_name}}",
"body": (
"Dear {{family_name}},\n\n"
"This is a reminder that the following service is scheduled for "
"tomorrow at {{cemetery_name}}:\n\n"
"Service Type: {{service_type}}\n"
"For: {{decedent_name}}\n"
"Date: {{service_date}}\n"
"Time: {{service_time}}\n"
"Location: {{location}}\n\n"
"If you need to make any changes, please contact us as soon as "
"possible at {{cemetery_email}} or {{cemetery_phone}}.\n\n"
"Warm regards,\n{{cemetery_name}}"
),
"merge_fields": [
"family_name", "service_type", "decedent_name", "service_date",
"service_time", "location",
],
},
{
"template_key": "news_notification",
"name": "News post notification",
"description": "Sent to subscribers when a news post is published",
"status": "active",
"subject": "New article: {{post_title}}",
"body": (
"{{post_title}}\n\n"
"{{post_excerpt}}\n\n"
"Read the full article: {{article_url}}\n\n"
"---\n"
"To unsubscribe from news notifications: {{unsubscribe_url}}"
),
"merge_fields": [
"post_title", "post_excerpt", "article_url", "unsubscribe_url",
],
},
]
def _sample_value_for(field: str) -> str:
"""Best-effort placeholder value for a merge field, used by send_test only."""
known = {
"cemetery_name": "Sample Cemetery",
"cemetery_email": "info@samplecemetery.org",
"cemetery_phone": "(555) 010-0100",
"current_date": datetime.now(timezone.utc).strftime("%B %d, %Y"),
}
return known.get(field, f"Sample {field.replace('_', ' ')}")
class EmailTemplateService:
@staticmethod
async def list_for_tenant(db: AsyncSession, tenant_id: UUID) -> list[EmailTemplate]:
result = await db.execute(
select(EmailTemplate)
.where(
EmailTemplate.tenant_id == tenant_id,
EmailTemplate.deleted_at.is_(None),
)
.order_by(EmailTemplate.name.asc())
)
return list(result.scalars().all())
@staticmethod
async def get_by_id(
db: AsyncSession, tenant_id: UUID, template_id: UUID
) -> Optional[EmailTemplate]:
result = await db.execute(
select(EmailTemplate).where(
EmailTemplate.id == template_id,
EmailTemplate.tenant_id == tenant_id,
EmailTemplate.deleted_at.is_(None),
)
)
return result.scalar_one_or_none()
@staticmethod
async def _write_audit(
db: AsyncSession,
tenant_id: UUID,
template_id: Optional[UUID],
action: str,
user_id: Optional[UUID],
ip_address: Optional[str] = None,
user_agent: Optional[str] = None,
old_value: Optional[dict] = None,
new_value: Optional[dict] = None,
) -> None:
audit = AuditLog(
tenant_id=tenant_id,
entity_type="email_template",
entity_id=template_id,
action=action,
old_value=old_value,
new_value=new_value,
ip_address=ip_address,
user_agent=user_agent,
user_id=user_id,
)
db.add(audit)
await db.flush()
@staticmethod
async def create(
db: AsyncSession,
tenant_id: UUID,
payload,
user_id: Optional[UUID],
ip_address: Optional[str] = None,
user_agent: Optional[str] = None,
) -> EmailTemplate:
existing = await db.execute(
select(EmailTemplate).where(
EmailTemplate.tenant_id == tenant_id,
EmailTemplate.deleted_at.is_(None),
EmailTemplate.name == payload.name,
)
)
if existing.scalar_one_or_none() is not None:
raise ConflictError("A template with this name already exists")
template = EmailTemplate(
tenant_id=tenant_id,
template_key=None,
is_system=False,
name=payload.name,
description=None,
subject=payload.subject,
body=payload.body,
status=payload.status,
merge_fields=list(_GLOBAL_MERGE_FIELDS),
)
db.add(template)
try:
await db.flush()
except IntegrityError as exc:
# Defense in depth: the SELECT pre-check above has a race window
# between two concurrent creates with the same name. The
# uq_email_templates_tenant_name partial unique index (migration
# 0048) is the safety net — translate the constraint violation
# back into the same 409 the pre-check raises.
await db.rollback()
raise ConflictError("A template with this name already exists") from exc
await EmailTemplateService._write_audit(
db,
tenant_id=tenant_id,
template_id=template.id,
action="create",
user_id=user_id,
ip_address=ip_address,
user_agent=user_agent,
new_value={
"name": template.name,
"subject": template.subject,
"body": template.body,
"status": template.status,
},
)
return template
@staticmethod
async def update(
db: AsyncSession,
tenant_id: UUID,
template_id: UUID,
payload,
user_id: Optional[UUID],
ip_address: Optional[str] = None,
user_agent: Optional[str] = None,
) -> EmailTemplate:
template = await EmailTemplateService.get_by_id(db, tenant_id, template_id)
if not template:
raise NotFoundError("Template not found")
update_data = payload.model_dump(exclude_unset=True)
# Defense in depth: is_system / template_key are immutable and must
# never be present in the schema-derived update dict (SEC-08 / API3).
assert "is_system" not in update_data
assert "template_key" not in update_data
if not update_data:
return template
# Duplicate-name check if name is being changed.
if "name" in update_data and update_data["name"] != template.name:
existing = await db.execute(
select(EmailTemplate).where(
EmailTemplate.tenant_id == tenant_id,
EmailTemplate.deleted_at.is_(None),
EmailTemplate.name == update_data["name"],
EmailTemplate.id != template_id,
)
)
if existing.scalar_one_or_none() is not None:
raise ConflictError("A template with this name already exists")
old_value = {
"subject": template.subject,
"body": template.body,
"status": template.status,
}
for field in ("name", "subject", "body", "status"):
if field in update_data:
setattr(template, field, update_data[field])
try:
await db.flush()
except IntegrityError as exc:
# Defense in depth: same race window as create() — the
# uq_email_templates_tenant_name unique index (migration 0048)
# is the safety net behind the SELECT-based pre-check above.
await db.rollback()
raise ConflictError("A template with this name already exists") from exc
new_value = {
"subject": template.subject,
"body": template.body,
"status": template.status,
}
await EmailTemplateService._write_audit(
db,
tenant_id=tenant_id,
template_id=template.id,
action="update",
user_id=user_id,
ip_address=ip_address,
user_agent=user_agent,
old_value=old_value,
new_value=new_value,
)
return template
@staticmethod
async def patch_status(
db: AsyncSession,
tenant_id: UUID,
template_id: UUID,
status: str,
user_id: Optional[UUID],
ip_address: Optional[str] = None,
user_agent: Optional[str] = None,
) -> EmailTemplate:
template = await EmailTemplateService.get_by_id(db, tenant_id, template_id)
if not template:
raise NotFoundError("Template not found")
old_status = template.status
template.status = status
await db.flush()
await EmailTemplateService._write_audit(
db,
tenant_id=tenant_id,
template_id=template.id,
action="status_change",
user_id=user_id,
ip_address=ip_address,
user_agent=user_agent,
old_value={"status": old_status},
new_value={"status": template.status},
)
return template
@staticmethod
async def soft_delete(
db: AsyncSession,
tenant_id: UUID,
template_id: UUID,
user_id: Optional[UUID],
ip_address: Optional[str] = None,
user_agent: Optional[str] = None,
) -> None:
template = await EmailTemplateService.get_by_id(db, tenant_id, template_id)
if not template:
raise NotFoundError("Template not found")
if template.is_system:
raise ForbiddenError("System templates cannot be deleted")
template.soft_delete()
await db.flush()
await EmailTemplateService._write_audit(
db,
tenant_id=tenant_id,
template_id=template.id,
action="delete",
user_id=user_id,
ip_address=ip_address,
user_agent=user_agent,
old_value={"status": template.status},
)
@staticmethod
async def send_test(
db: AsyncSession,
tenant_id: UUID,
template_id: UUID,
requesting_user: User,
to_email: Optional[str] = None,
ip_address: Optional[str] = None,
user_agent: Optional[str] = None,
) -> str:
"""Send a test render of the template. Returns the recipient address used.
`to_email` may target any address (product decision) — the requester's
own email is the default when omitted. Abuse is bounded by the
5/hour-per-(user, template) rate limit and audit logging in the
router, not by restricting the recipient.
"""
template = await EmailTemplateService.get_by_id(db, tenant_id, template_id)
if not template:
raise NotFoundError("Template not found")
recipient = to_email or requesting_user.email
rendered_subject = template.subject
rendered_body = template.body
for field in template.merge_fields or []:
sample_value = _sample_value_for(field)
placeholder = "{{" + field + "}}"
rendered_subject = rendered_subject.replace(placeholder, sample_value)
rendered_body = rendered_body.replace(placeholder, sample_value)
# NOTE: AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY are intentionally NOT
# treated as "configured" here. src/core/email.py's _dispatch() has no
# SES/boto3 branch yet — it only sends via SMTP when SMTP_HOST is set,
# otherwise it silently logs to console and returns without error.
# Counting AWS keys alone would report success while sending nothing
# (see CLAUDE.md "Known Technical Debt" — SES email templates missing).
email_configured = bool(settings.SMTP_HOST)
outcome = "sent" if email_configured else "failed_not_configured"
try:
if not email_configured:
raise ServiceUnavailableError("Email service not configured")
# Template bodies are stored/rendered as plain text (merge fields
# are substituted via plain str.replace — never a real HTML
# authoring UI or template engine). Passing rendered_body
# unescaped as body_html would let any content that slips past
# the schema-level denylist execute as live HTML in the
# recipient's mail client. HTML-escape it and convert newlines
# to
for readability; body_text stays the plain, unescaped
# rendered_body (safe as-is inside the MIMEText "plain" part).
escaped_body_html = html.escape(rendered_body).replace("\n", "
\n")
await _dispatch(
to=recipient,
subject=rendered_subject,
body_text=rendered_body,
body_html=escaped_body_html,
)
finally:
await EmailTemplateService._write_audit(
db,
tenant_id=tenant_id,
template_id=template.id,
action="test_send",
user_id=requesting_user.id,
ip_address=ip_address,
user_agent=user_agent,
new_value={"outcome": outcome, "to": recipient},
)
return recipient
@staticmethod
async def get_by_trigger_key(
db: AsyncSession, tenant_id: UUID, trigger_key: str
) -> Optional[EmailTemplate]:
"""Resolve a tenant's system template for a dispatch trigger (INDL-54).
Tenant-scoped (SEC A01) and filters soft-deleted rows. Returns None
when the tenant has no row for this trigger — the dispatch engine then
self-heals via `seed_missing_system_template`.
"""
result = await db.execute(
select(EmailTemplate).where(
EmailTemplate.tenant_id == tenant_id,
EmailTemplate.template_key == trigger_key,
EmailTemplate.deleted_at.is_(None),
)
)
return result.scalar_one_or_none()
@staticmethod
async def seed_missing_system_template(
db: AsyncSession, tenant_id: UUID, spec: dict
) -> Optional[EmailTemplate]:
"""Insert one missing system template row from a default spec (INDL-54).
Used by the dispatch engine's self-heal path (Lookup precedence case 3)
so a legacy/unseeded tenant gets an editable copy after the first send.
Uses INSERT ... ON CONFLICT DO NOTHING so a concurrent seed can never
raise an IntegrityError that would poison the caller's transaction
(fail-open, AC-12). Returns the resulting row (freshly inserted or the
pre-existing one on conflict), or None if it could not be resolved.
"""
from sqlalchemy.dialects.postgresql import insert as pg_insert
stmt = (
pg_insert(EmailTemplate.__table__)
.values(
tenant_id=tenant_id,
template_key=spec["template_key"],
is_system=True,
name=spec["name"],
description=spec.get("description"),
subject=spec["subject"],
body=spec["body"],
status=spec.get("status", "active"),
merge_fields=list(spec["merge_fields"]) + list(_GLOBAL_MERGE_FIELDS),
)
.on_conflict_do_nothing()
)
# Run inside a SAVEPOINT so any unexpected flush failure rolls back only
# this nested block and never poisons the caller's outer transaction
# (fail-open, AC-12). ON CONFLICT DO NOTHING already prevents the normal
# duplicate-key path from raising.
async with db.begin_nested():
await db.execute(stmt)
return await EmailTemplateService.get_by_trigger_key(
db, tenant_id, spec["template_key"]
)
@staticmethod
async def seed_system_templates(db: AsyncSession, tenant_id: UUID) -> list[EmailTemplate]:
"""Idempotently insert the 17 system templates for one tenant.
Skips any template_key that already exists for the tenant (including
soft-deleted rows, since template_key has a partial-unique index and
we don't want to reintroduce a duplicate). Used both by new-tenant
provisioning and directly importable for tests / backfill scripts.
"""
existing_result = await db.execute(
select(EmailTemplate.template_key).where(
EmailTemplate.tenant_id == tenant_id,
EmailTemplate.template_key.is_not(None),
)
)
existing_keys = {row[0] for row in existing_result.all()}
created: list[EmailTemplate] = []
for spec in SYSTEM_TEMPLATES:
if spec["template_key"] in existing_keys:
continue
template = EmailTemplate(
tenant_id=tenant_id,
template_key=spec["template_key"],
is_system=True,
name=spec["name"],
description=spec["description"],
subject=spec["subject"],
body=spec["body"],
status=spec["status"],
merge_fields=list(spec["merge_fields"]) + list(_GLOBAL_MERGE_FIELDS),
)
db.add(template)
created.append(template)
if created:
await db.flush()
return created