第 9 課：模板引擎和內容管道

設計一個通知系統來發送數百萬封電子郵件

第 3 部分：電子郵件基礎設施和交付引擎

亞洲開發網

簡介

當發送 1000 萬封電子郵件時，每封電子郵件可能需要不同的內容 - 名稱、建議產品、唯一折扣代碼。模板引擎是負責快速和個性化呈現數百萬封電子郵件的組件。

---

1. 電子郵件範本技術

比較

技術	文法	輸出	反應靈敏	學習曲線
MJML	類別 XML	HTML 電子郵件	✅ 內建	簡單
車把	{{var}}	HTML/文字	手冊	簡單
反應電子郵件	JSX	HTML 電子郵件	✅ 元件	中等
液體	{{var}}	HTML/文字	手冊	簡單
電子JS	<%= var %>	HTML/文字	手冊	簡單

MJML 範例

<mjml>
  <mj-head>
    <mj-attributes>
      <mj-all font-family="Helvetica, Arial, sans-serif" />
    </mj-attributes>
  </mj-head>
  <mj-body background-color="#f4f4f4">
    <mj-section background-color="#ffffff" padding="20px">
      <mj-column>
        <mj-image src="https://cdn.shop.com/logo.png" width="150px" />
        <mj-text font-size="24px" color="#333">
          Xin chào {{first_name}}! 🎉
        </mj-text>
        <mj-text>
          Flash Sale đang diễn ra! Sử dụng mã
          <strong>{{discount_code}}</strong> để được giảm
          <strong>{{discount_percent}}%</strong>.
        </mj-text>
        <mj-button background-color="#e74c3c" href="{{cta_url}}">
          Mua ngay
        </mj-button>
      </mj-column>
    </mj-section>

    <!-- Personalized product recommendations -->
    {{#each recommended_products}}
    <mj-section>
      <mj-column>
        <mj-image src="{{this.image_url}}" />
        <mj-text>{{this.name}} - {{this.price}}</mj-text>
        <mj-button href="{{this.url}}">Xem chi tiết</mj-button>
      </mj-column>
    </mj-section>
    {{/each}}

    <mj-section>
      <mj-column>
        <mj-text font-size="12px" color="#999">
          <a href="{{unsubscribe_url}}">Hủy đăng ký</a>
        </mj-text>
      </mj-column>
    </mj-section>
  </mj-body>
</mjml>

---

2. 內容管道架構

┌────────────┐    ┌────────────┐    ┌────────────┐    ┌────────────┐
│  Template   │───▶│   Render   │───▶│  Validate  │───▶│   Send     │
│  Storage    │    │   Engine   │    │  & Inline  │    │  to ESP    │
│            │    │            │    │            │    │            │
│ MJML/HTML  │    │ + Context  │    │ CSS inline │    │ Final HTML │
│ Templates  │    │ = HTML     │    │ Image URLs │    │            │
└────────────┘    └────────────┘    └────────────┘    └────────────┘

實作

class ContentPipeline:
    def __init__(self):
        self.template_store = TemplateStore()    # Redis + DB
        self.mjml_compiler = MJMLCompiler()       # MJML → HTML
        self.handlebars = HandlebarsEngine()       # Variable substitution
        self.css_inliner = CSSInliner()           # Inline CSS
        self.link_tracker = LinkTracker()          # Tracking links

    async def render(
        self, template_id: str, context: dict
    ) -> RenderedEmail:
        # 1. Get compiled template (cached)
        template = await self.get_compiled_template(template_id)

        # 2. Render variables
        html = self.handlebars.render(template.html, context)
        subject = self.handlebars.render(template.subject, context)

        # 3. Inline CSS (email clients strip <style> tags)
        html = self.css_inliner.inline(html)

        # 4. Add tracking
        html = self.link_tracker.wrap_links(
            html,
            campaign_id=context.get('campaign_id'),
            recipient_id=context.get('recipient_id'),
        )

        # 5. Add tracking pixel
        html = self.add_open_tracking_pixel(
            html,
            message_id=context.get('message_id'),
        )

        # 6. Generate text version
        text = self.html_to_text(html)

        return RenderedEmail(
            subject=subject,
            html=html,
            text=text,
        )

    async def get_compiled_template(self, template_id: str):
        # Check cache first
        cached = await self.redis.get(f"template:{template_id}")
        if cached:
            return CompiledTemplate.from_cache(cached)

        # Compile MJML → HTML
        template = await self.template_store.get(template_id)
        compiled_html = self.mjml_compiler.compile(template.mjml_source)

        # Cache for 1 hour
        compiled = CompiledTemplate(
            html=compiled_html,
            subject=template.subject,
        .(version=template.version,
        )
        await self.redis.set(//
            f"template:{template_id}",
            compiled.to_cache(),
            ex=3600
        )
        return compiled

---

3. 模板版本控制和 A/B 測試

版本管理

class TemplateVersionManager:
    async def create_version(
        self, template_id: str, mjml_source: str, subject: str
    ):
        current = await self.db.get_active_version(template_id)
        new_version = (current.version if current else 0) + 1

        await self.db.insert('template_versions', {
            'template_id': template_id,
            'version': new_version,
            'mjml_source	': mjml_source,
            'subject': subject,
            'is_active': False,  ishi# Not active until published
            'created_at': datetime.utcnow(),
        })

        # Invalidate cache
        await self.redis.delete(f"template:{template_id	}")

    async def publish_version(self, template_id: str, version: int):
        await self.db.execute(
            "UPDATE template_versions SET is_active = false "
            "WHERE template_id = %s", [template_id]
        )
        await self.db.execute(
            "UPDATE template_versions SET is_active = true "
            "WHERE template_id = %s AND version = %s",
            [template_id, version]
        )

A/B 測試

class ABTestingEngine:
    async def create_ab_test(
        self,
        campaign_id: str,
        variants: list[dict],
        split_ratio: list[float],
    ):
        """
        variants = [
            {'template_id': 'tmpl_v1', 'subject': 'Flash Sale!'},
            {'template_id': 'tmpl_v2', 'subject': '30% OFF Today Only'},
        ]
        split_ratio = [0.5, 0.5]  # 50/50 split
        """
        await self.db.insert('ab_tests', {
            'campaign_id': campaign_id,
            'variants': variants,
            'split_ratio': split_ratio,
            'winner': None,
            'status': 'running',
        })

    def assign_variant(self, recipient_id: str, ab_test: dict) -> dict:
        """Deterministic variant assignment based on recipient ID"""
        hash_value = hash(recipient_id) % 100
        cumulative = 0
        for i, ratio in enumerate(ab_test['split_ratio']):
            cumulative += ratio * 100
            if hash_value < cumulative:
                return ab_test['variants'][i]
        return ab_test['variants'][-1]

    async def determine_winner(self, campaign_id: str):
        """After campaign completes, determine winning variant"""
        results = await self.db.query(
            "SELECT variant_id, "
            "  COUNT(*) as sent, "
            "  SUM(CASE WHEN opened_at IS NOT NULL THEN 1 END) as opens, "
            "  SUM(CASE WHEN clicked_at IS NOT NULL THEN 1 END) as clicks "
            "FROM email_messages "
            "WHERE campaign_id = %s "
            "GROUP BY variant_id",
            [campaign_id]
        )
        # Winner = highest click rate
        winner = max(results, key=lambda r: r['clicks'] / r['sent'])
        return winner

---

4. 預渲染與快取策略

問題

1000 萬封電子郵件 × 50 毫秒渲染 = 139 小時 渲染時間 😱

解決方案 1：在背景預渲染

class PreRenderService:
    async def pre_render_campaign(self, campaign_id: str):
        """Pre-render all emails before campaign starts"""
        campaign = await self.get_campaign(campaign_id)
        template = await self.get_template(campaign.template_id)

        async for batch in self.stream_recipients(campaign, batch_size=500):
            rendered_batch = []
            for recipient in batch:
                context = self.build_context(campaign, recipient)
                rendered = await self.pipeline.render(
                    campaign.template_id, context
                )
                rendered_batch.append({
                    'message_id': generate_id(),
                    'recipient': recipient.email,
                    'subject': rendered.subject,
                    'html': rendered.html,
                })

            # Store pre-rendered emails
            await self.store_rendered_batch(campaign_id, rendered_batch)

        await self.update_campaign_status(campaign_id, 'PRE_RENDERED')

解決方案2：模板片段緩存

class FragmentCachingPipeline:
    """Cache static parts, only render dynamic parts"""

    async def render_with_cache(self, template_id: str, context: dict):
        # Static parts (header, footer, layout) → cached
        static_html = await self.get_static_parts(template_id)

        # Dynamic parts only → render per recipient
        dynamic_html = self.render_dynamic_slots(static_html, context)

        return dynamic_html

    async def get_static_parts(self, template_id: str):
        cache_key = f"template_static:{template_id}"
        cached = await self.redis.get(cache_key)
        if cached:
            return cached

        template = await self.template_store.get(template_id)
        # Compile MJML, inline CSS (these are expensive)
        static = self.compile_and_inline(template)
        await self.redis.set(cache_key, static, ex=3600)
        return static

效能比較

Full render:      50ms per email → 139 hours for 10M
Fragment cache:   5ms per email  → 13.9 hours for 10M
Pre-render:       0ms at send time → instant (pre-computed)

With 20 workers:
Fragment cache:   13.9 / 20 = ~42 minutes
Pre-render:       13.9 / 20 = ~42 min prep, 0 at send time

---

5.動態內容個人化

個性化變數

class PersonalizationEngine:
    async def build_context(
        self, campaign: Campaign, recipient: Recipient
    ) -> dict:
        return {
            # Basic info
            'first_name': recipient.first_name or 'bạn',
            'email': recipient.email,

            # Campaign specific
            'discount_code': await self.generate_unique_code(recipient.id),
            'discount_percent': campaign.metadata['discount'],

            # Personalized recommendations
            'recommended_products': await self.get_recommendations(
                recipient.id, limit=3
            ),

            # Dynamic URLs
            'cta_url': self.build_tracked_url(
                campaign.cta_url,
                campaign_id=campaign.id,
                recipient_id=recipient.id,
            ),
            'unsubscribe_url': self.build_unsubscribe_url(
                recipient.email,
                campaign.id,
            ),

            # Conditional content
            'is_vip': recipient.tier == 'vip',
            'locale': recipient.locale or 'vi',
        }

範本中的條件內容

{{#if is_vip}}
  <mj-section background-color="#ffd700">
    <mj-text>⭐ Ưu đãi VIP: Giảm thêm 10%!</mj-text>
  </mj-section>
{{/if}}

{{#if (eq locale 'en')}}
  <mj-text>Hello {{first_name}}!</mj-text>
{{else}}
  <mj-text>Xin chào {{first_name}}!</mj-text>
{{/if}}

---

6. 合規性 — 取消訂閱和反垃圾郵件

一鍵取消訂閱 (RFC 8058)

class UnsubscribeManager:
    def build_headers(self, email: str, campaign_id: str) -> dict:
        token = self.generate_unsubscribe_token(email, campaign_id)
        return {
            'List-Unsubscribe': (
                f'<https://mail.yourdomain.com/unsubscribe?token={token}>, '
                f'<mailto:[email protected]?subject={token}>'
            ),
            'List-Unsubscribe-Post': 'List-Unsubscribe=One-Click',
        }

    async def handle_unsubscribe(self, token: str):
        data = self.verify_token(token)
        await self.suppression_service.add(
            email=data['email'],
            reason='unsubscribe',
            campaign_id=data['campaign_id'],
        )

---

總結

• MJML 是響應式電子郵件範本的最佳選擇
• 內容管道：範本→渲染→CSS內聯→追蹤→發送
• 預先渲染或片段快取可將渲染時間減少 10 倍
• A/B 測試 有助於優化開啟率和點擊率
• 始終遵守 CAN-SPAM，一鍵取消訂閱

下一篇文章： 速率限制與節流 — 控制電子郵件傳送速度。