Payload CMS Pub Sub Architecture

Payload CMS Pub/Sub

Payload CMS Headless TypeScript React Pub/Sub Architecture Event-Driven Kafka RabbitMQ Redis Hooks REST GraphQL MongoDB PostgreSQL Real-time Content

Payload CMS Setup และ Hooks

# === Payload CMS with Pub/Sub ===

# npx create-payload-app@latest my-cms
# cd my-cms && npm run dev

# payload.config.ts
# import { buildConfig } from 'payload/config'
# import { mongooseAdapter } from '@payloadcms/db-mongodb'
#
# export default buildConfig({
#   db: mongooseAdapter({ url: process.env.MONGODB_URI }),
#   collections: [
#     {
#       slug: 'articles',
#       fields: [
#         { name: 'title', type: 'text', required: true },
#         { name: 'content', type: 'richText' },
#         { name: 'category', type: 'select',
#           options: ['tech', 'finance', 'lifestyle'] },
#         { name: 'status', type: 'select',
#           options: ['draft', 'published', 'archived'] },
#         { name: 'author', type: 'relationship', relationTo: 'users' },
#       ],
#       hooks: {
#         afterChange: [publishEvent],
#         afterDelete: [publishDeleteEvent],
#       },
#     },
#   ],
# })
#
# // Hook: Publish Event
# async function publishEvent({ doc, operation, req }) {
#   const event = {
#     type: `article.`,  // created or updated
#     payload: { id: doc.id, title: doc.title, status: doc.status },
#     timestamp: new Date().toISOString(),
#   }
#   await redis.publish('content-events', JSON.stringify(event))
#   console.log(`Published: `)
# }

from dataclasses import dataclass
from typing import List

@dataclass
class ContentEvent:
    event_type: str
    collection: str
    doc_id: str
    title: str
    timestamp: str

events = [
    ContentEvent("article.created", "articles", "a001", "วิธีใช้ Docker", "10:00:00"),
    ContentEvent("article.updated", "articles", "a002", "Python Tips", "10:05:00"),
    ContentEvent("article.published", "articles", "a001", "วิธีใช้ Docker", "10:10:00"),
    ContentEvent("page.created", "pages", "p001", "About Us", "10:15:00"),
    ContentEvent("article.deleted", "articles", "a003", "Old Post", "10:20:00"),
]

print("=== Content Events ===")
for e in events:
    print(f"  [{e.timestamp}] {e.event_type}")
    print(f"    Collection: {e.collection} | ID: {e.doc_id} | Title: {e.title}")

Pub/Sub Implementation

# === Pub/Sub with Redis ===

# Publisher (Payload Hook)
# import Redis from 'ioredis'
# const redis = new Redis(process.env.REDIS_URL)
#
# async function publishEvent(channel, event) {
#   await redis.publish(channel, JSON.stringify(event))
# }
#
# // Subscriber (Search Indexer)
# const subscriber = new Redis(process.env.REDIS_URL)
# subscriber.subscribe('content-events')
# subscriber.on('message', async (channel, message) => {
#   const event = JSON.parse(message)
#   if (event.type === 'article.created' || event.type === 'article.updated') {
#     await elasticsearch.index({
#       index: 'articles',
#       id: event.payload.id,
#       body: event.payload,
#     })
#   }
# })

# Kafka Alternative
# from kafka import KafkaProducer, KafkaConsumer
# import json
#
# producer = KafkaProducer(
#     bootstrap_servers=['localhost:9092'],
#     value_serializer=lambda v: json.dumps(v).encode('utf-8'),
# )
#
# def publish_content_event(event):
#     producer.send('content-events', value=event)
#     producer.flush()
#
# consumer = KafkaConsumer(
#     'content-events',
#     bootstrap_servers=['localhost:9092'],
#     value_deserializer=lambda m: json.loads(m.decode('utf-8')),
#     group_id='search-indexer',
# )
#
# for message in consumer:
#     event = message.value
#     handle_event(event)

@dataclass
class Subscriber:
    name: str
    topic: str
    action: str
    latency: str

subscribers = [
    Subscriber("Search Indexer", "content-events", "Update Elasticsearch", "< 1s"),
    Subscriber("Cache Invalidator", "content-events", "Clear CDN Cache", "< 2s"),
    Subscriber("SSG Rebuilder", "content-events", "Trigger Next.js ISR", "< 5s"),
    Subscriber("Notification", "content-events", "Send Push Notification", "< 3s"),
    Subscriber("Analytics", "content-events", "Log Content Changes", "< 1s"),
    Subscriber("Backup Service", "content-events", "Backup to S3", "< 10s"),
]

print("\n=== Pub/Sub Subscribers ===")
for s in subscribers:
    print(f"  [{s.name}] Topic: {s.topic}")
    print(f"    Action: {s.action} | Latency: {s.latency}")

Production Architecture

# === Production Setup ===

# docker-compose.yml
# version: '3.8'
# services:
#   payload:
#     build: .
#     ports: ["3000:3000"]
#     environment:
#       - MONGODB_URI=mongodb://mongo:27017/cms
#       - REDIS_URL=redis://redis:6379
#     depends_on: [mongo, redis]
#   mongo:
#     image: mongo:7
#     volumes: [mongo-data:/data/db]
#   redis:
#     image: redis:7-alpine
#     ports: ["6379:6379"]
#   search-indexer:
#     build: ./services/search-indexer
#     environment:
#       - REDIS_URL=redis://redis:6379
#       - ES_URL=http://elasticsearch:9200
#   elasticsearch:
#     image: elasticsearch:8.12.0
#     environment:
#       - discovery.type=single-node
#     ports: ["9200:9200"]

architecture = {
    "Payload CMS": "Content Management + REST/GraphQL API",
    "MongoDB": "Content Storage (Primary DB)",
    "Redis": "Pub/Sub Message Broker + Cache",
    "Elasticsearch": "Full-text Search Index",
    "Next.js": "Frontend SSG/ISR",
    "CDN": "Static Asset Delivery",
    "S3": "Media Upload Storage",
}

print("Architecture Components:")
for component, role in architecture.items():
    print(f"  [{component}]: {role}")

# Event Flow
flow = [
    "1. Editor สร้าง/แก้ Content ใน Payload Admin",
    "2. afterChange Hook Publish Event ไป Redis",
    "3. Search Indexer รับ Event อัพเดท Elasticsearch",
    "4. Cache Invalidator ล้าง CDN Cache",
    "5. SSG Rebuilder Trigger ISR สร้างหน้าใหม่",
    "6. Notification Service ส่ง Push ให้ Subscribers",
    "7. Frontend แสดงเนื้อหาใหม่ภายใน 5 วินาที",
]

print(f"\n\nEvent Flow:")
for step in flow:
    print(f"  {step}")

เคล็ดลับ

• Hooks: ใช้ afterChange Hook ไม่ใช่ beforeChange สำหรับ Pub
• Idempotent: Subscriber ต้อง Idempotent รับ Event ซ้ำได้
• Dead Letter: ตั้ง Dead Letter Queue สำหรับ Event ที่ Process ไม่ได้
• Retry: Subscriber ต้องมี Retry Logic กรณี External Service Down
• Monitoring: Monitor Queue Size และ Consumer Lag

การนำความรู้ไปประยุกต์ใช้งานจริง

แหล่งเรียนรู้ที่แนะนำ ได้แก่ Official Documentation ที่อัพเดทล่าสุดเสมอ Online Course จาก Coursera Udemy edX ช่อง YouTube คุณภาพทั้งไทยและอังกฤษ และ Community อย่าง Discord Reddit Stack Overflow ที่ช่วยแลกเปลี่ยนประสบการณ์กับนักพัฒนาทั่วโลก

สรุป

Payload CMS Headless TypeScript Pub/Sub Architecture Redis Kafka Event-Driven Hooks afterChange Search Cache ISR Notification Microservices MongoDB Elasticsearch Production