"""create site_templates — replace notification_templates PRD-010 Templates Management. Actions: 1. Creates `site_templates` with all columns, constraints, and indexes. 2. Copies matching rows from `notification_templates` with key remapping. 3. Drops `notification_templates`. Key mapping (old → new): hpp_payment_request (email) → hpp_payment_request payment_receipt (email) → transaction_receipt_email customer_verify (email) → account_verification payment_failed_retry (email) → payment_failed customer_account_created → merchant_welcome Revision ID: 20260321120000 Revises: f8a9b0c1d2e3 Create Date: 2026-03-21 12:00:00.000000 """ from typing import Sequence, Union from alembic import op import sqlalchemy as sa from sqlalchemy import inspect as sa_inspect, text # --------------------------------------------------------------------------- # Revision metadata # --------------------------------------------------------------------------- revision: str = "20260321120000" down_revision: Union[str, Sequence[str], None] = "f8a9b0c1d2e3" branch_labels: Union[str, Sequence[str], None] = None depends_on: Union[str, Sequence[str], None] = None # --------------------------------------------------------------------------- # upgrade # --------------------------------------------------------------------------- def upgrade() -> None: conn = op.get_bind() existing_tables = sa_inspect(conn).get_table_names() # ------------------------------------------------------------------ # 1. CREATE TABLE site_templates # ------------------------------------------------------------------ if "site_templates" not in existing_tables: op.create_table( "site_templates", sa.Column( "id", sa.Integer(), autoincrement=True, nullable=False, comment="Surrogate primary key", ), sa.Column( "template_key", sa.String(100), nullable=False, comment="Machine-readable key used to look up the template at render time", ), sa.Column( "template_name", sa.String(255), nullable=False, comment="Human-readable display name shown in the admin UI", ), sa.Column( "description", sa.Text(), nullable=True, comment="Optional admin-facing description of when this template is sent", ), sa.Column( "channel", sa.String(10), nullable=False, comment="Delivery channel: email | sms | pdf", ), sa.Column( "template_type", sa.String(20), nullable=False, comment="Functional category: auth | payment | user_invite", ), sa.Column( "subject", sa.String(255), nullable=True, comment="Email subject line; null for SMS and PDF templates", ), sa.Column( "body_html", sa.Text(), nullable=True, comment="Full HTML body; null for SMS-only templates", ), sa.Column( "body_text", sa.Text(), nullable=True, comment="Plain-text body; used as SMS message and email plain-text fallback", ), sa.Column( "primary_color", sa.String(7), nullable=False, server_default="#28B4ED", comment="Hex color injected as {{ primary_color }} in payment/invite templates", ), sa.Column( "accent_color", sa.String(7), nullable=False, server_default="#FB7585", comment="Hex color injected as {{ accent_color }} in payment/invite templates", ), sa.Column( "text_color", sa.String(7), nullable=False, server_default="#252525", comment="Hex color injected as {{ text_color }} in payment/invite templates", ), sa.Column( "variables", sa.JSON(), nullable=False, server_default="[]", comment="Ordered list of Jinja2 variable names expected by this template", ), sa.Column( "is_active", sa.Boolean(), nullable=False, server_default="true", comment="Inactive templates are skipped by the notification dispatcher", ), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("now()"), comment="Row creation timestamp (UTC)", ), sa.Column( "updated_at", sa.DateTime(timezone=True), nullable=True, comment="Last modification timestamp; null until first update", ), sa.Column( "deleted_at", sa.DateTime(timezone=True), nullable=True, comment="Soft-delete timestamp; non-null rows are excluded from all queries", ), sa.PrimaryKeyConstraint("id", name="pk_site_templates"), sa.UniqueConstraint("template_key", name="uq_site_templates_template_key"), comment=( "Platform-level template store (PRD-010). Replaces notification_templates. " "No merchant_id — all templates are shared platform defaults rendered with " "per-merchant branding injected at runtime." ), ) op.create_index( "ix_site_templates_id", "site_templates", ["id"], unique=False, ) op.create_index( "ix_site_templates_channel", "site_templates", ["channel"], unique=False, ) op.create_index( "ix_site_templates_template_type", "site_templates", ["template_type"], unique=False, ) # ------------------------------------------------------------------ # 2. Copy matching rows from notification_templates with key remapping # Only copies rows where merchant_id IS NULL (platform defaults). # channel 'sms' rows from the old table are not carried over here # because new sms keys have been renamed; they will be seeded fresh. # ------------------------------------------------------------------ op.execute( text( """ INSERT INTO site_templates ( template_key, template_name, description, channel, template_type, subject, body_html, body_text, primary_color, accent_color, text_color, variables, is_active, created_at, updated_at, deleted_at ) SELECT CASE nt.template_key WHEN 'hpp_payment_request' THEN 'hpp_payment_request' WHEN 'payment_receipt' THEN 'transaction_receipt_email' WHEN 'customer_verify' THEN 'account_verification' WHEN 'payment_failed_retry' THEN 'payment_failed' WHEN 'customer_account_created' THEN 'merchant_welcome' END AS template_key, CASE nt.template_key WHEN 'hpp_payment_request' THEN 'Payment Request (HPP)' WHEN 'payment_receipt' THEN 'Transaction Receipt (Email)' WHEN 'customer_verify' THEN 'Account Verification' WHEN 'payment_failed_retry' THEN 'Payment Failed' WHEN 'customer_account_created' THEN 'Merchant Welcome' END AS template_name, NULL AS description, nt.channel AS channel, CASE nt.template_key WHEN 'hpp_payment_request' THEN 'payment' WHEN 'payment_receipt' THEN 'payment' WHEN 'customer_verify' THEN 'auth' WHEN 'payment_failed_retry' THEN 'payment' WHEN 'customer_account_created' THEN 'auth' END AS template_type, nt.subject AS subject, nt.body_html AS body_html, nt.body_text AS body_text, '#28B4ED' AS primary_color, '#FB7585' AS accent_color, '#252525' AS text_color, '[]'::json AS variables, nt.is_active AS is_active, nt.created_at AS created_at, nt.updated_at AS updated_at, nt.deleted_at AS deleted_at FROM notification_templates nt WHERE nt.merchant_id IS NULL AND nt.channel = 'email' AND nt.template_key IN ( 'hpp_payment_request', 'payment_receipt', 'customer_verify', 'payment_failed_retry', 'customer_account_created' ) AND nt.deleted_at IS NULL ON CONFLICT (template_key) DO NOTHING """ ) ) # ------------------------------------------------------------------ # 3. Drop notification_templates # ------------------------------------------------------------------ op.drop_index( "ix_notification_templates_merchant_key_channel", table_name="notification_templates", if_exists=True, ) op.drop_index( "ix_notification_templates_key_channel", table_name="notification_templates", if_exists=True, ) if "notification_templates" in existing_tables: op.drop_table("notification_templates") # --------------------------------------------------------------------------- # downgrade # WARNING: downgrade recreates schema only — data not restored # --------------------------------------------------------------------------- def downgrade() -> None: # ------------------------------------------------------------------ # 1. Recreate notification_templates (schema only — data not restored) # ------------------------------------------------------------------ op.create_table( "notification_templates", sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), sa.Column( "merchant_id", sa.Integer(), sa.ForeignKey("merchants.id", name="fk_notification_templates_merchant_id"), nullable=True, ), sa.Column("template_key", sa.String(100), nullable=False), sa.Column("channel", sa.String(10), nullable=False), sa.Column("subject", sa.String(255), nullable=True), sa.Column("body_html", sa.Text(), nullable=True), sa.Column("body_text", sa.Text(), nullable=True), sa.Column( "is_active", sa.Boolean(), nullable=False, server_default="true", ), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("now()"), ), sa.Column("updated_at", sa.DateTime(timezone=True), nullable=True), sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True), sa.PrimaryKeyConstraint("id", name="pk_notification_templates"), sa.UniqueConstraint( "merchant_id", "template_key", "channel", name="uq_notification_template_merchant_key_channel", ), ) op.create_index( "ix_notification_templates_key_channel", "notification_templates", ["template_key", "channel"], unique=False, ) op.create_index( "ix_notification_templates_merchant_key_channel", "notification_templates", ["merchant_id", "template_key", "channel"], unique=False, ) # ------------------------------------------------------------------ # 2. Drop site_templates # ------------------------------------------------------------------ op.drop_index("ix_site_templates_template_type", table_name="site_templates") op.drop_index("ix_site_templates_channel", table_name="site_templates") op.drop_index("ix_site_templates_id", table_name="site_templates") op.drop_table("site_templates")