WordPress Headless Pub Sub Architecture

WordPress Headless Pub/Sub

Headless WordPress Backend CMS REST API WPGraphQL Next.js React Frontend Pub/Sub Publish Subscribe Event-driven Decouple Scale Real-time Content Delivery

Headless WordPress Setup

# === Headless WordPress + WPGraphQL ===

# Install WPGraphQL Plugin
# wp plugin install wp-graphql --activate

# GraphQL Query — Get Posts
# query GetPosts {
#   posts(first: 10) {
#     nodes {
#       id
#       title
#       slug
#       date
#       excerpt
#       content
#       featuredImage {
#         node {
#           sourceUrl
#           altText
#         }
#       }
#       categories {
#         nodes {
#           name
#           slug
#         }
#       }
#     }
#   }
# }

# Next.js — Fetch Posts
# // lib/wordpress.ts
# const API_URL = process.env.WORDPRESS_API_URL;
#
# export async function getPosts() {
#   const res = await fetch(API_URL, {
#     method: 'POST',
#     headers: { 'Content-Type': 'application/json' },
#     body: JSON.stringify({
#       query: `
#         query GetPosts {
#           posts(first: 10) {
#             nodes {
#               id title slug date excerpt
#               featuredImage { node { sourceUrl } }
#             }
#           }
#         }
#       `,
#     }),
#     next: { revalidate: 60 },
#   });
#   const json = await res.json();
#   return json.data.posts.nodes;
# }
#
# // app/blog/page.tsx
# export default async function BlogPage() {
#   const posts = await getPosts();
#   return (
#     

#       {posts.map(post => (
#         

#           
{post.title}
#           
{post.excerpt}
#         
#       ))}
#     
#   );
# }

from dataclasses import dataclass, field
from typing import List

@dataclass
class HeadlessConfig:
    wordpress_url: str
    api_type: str  # rest, graphql
    frontend: str
    hosting: str
    cdn: str
    features: List[str]

config = HeadlessConfig(
    wordpress_url="https://cms.example.com",
    api_type="WPGraphQL",
    frontend="Next.js 14 (App Router)",
    hosting="Vercel",
    cdn="Vercel Edge Network",
    features=["ISR", "On-demand Revalidation", "Image Optimization",
              "Pub/Sub Webhooks", "Preview Mode"],
)

print("=== Headless WordPress Config ===")
print(f"  WordPress: {config.wordpress_url}")
print(f"  API: {config.api_type}")
print(f"  Frontend: {config.frontend}")
print(f"  Hosting: {config.hosting}")
print(f"  CDN: {config.cdn}")
print(f"  Features: {', '.join(config.features)}")

Pub/Sub Event System

# === Pub/Sub Architecture for WordPress ===

# WordPress Webhook Plugin (functions.php)
# function publish_event_on_save($post_id, $post, $update) {
#     if ($post->post_status !== 'publish') return;
#     if (wp_is_post_revision($post_id)) return;
#
#     $event = [
#         'event' => $update ? 'post.updated' : 'post.created',
#         'post_id' => $post_id,
#         'slug' => $post->post_name,
#         'title' => $post->post_title,
#         'type' => $post->post_type,
#         'timestamp' => current_time('c'),
#     ];
#
#     // Google Pub/Sub
#     wp_remote_post('https://pubsub.googleapis.com/v1/projects/myproject/topics/wordpress-events:publish', [
#         'headers' => ['Authorization' => 'Bearer ' . get_gcp_token()],
#         'body' => json_encode(['messages' => [['data' => base64_encode(json_encode($event))]]]),
#     ]);
# }
# add_action('save_post', 'publish_event_on_save', 10, 3);

# Python Pub/Sub Subscriber
# from google.cloud import pubsub_v1
# import json
# import requests
#
# subscriber = pubsub_v1.SubscriberClient()
# subscription_path = "projects/myproject/subscriptions/frontend-revalidate"
#
# def callback(message):
#     event = json.loads(message.data.decode('utf-8'))
#     print(f"Event: {event['event']} | Slug: {event['slug']}")
#
#     if event['event'] in ['post.created', 'post.updated']:
#         # Revalidate Next.js page
#         requests.post(
#             f"https://mysite.com/api/revalidate",
#             json={"slug": event['slug'], "secret": REVALIDATE_SECRET}
#         )
#
#         # Purge CDN cache
#         requests.post(
#             f"https://api.cloudflare.com/client/v4/zones/{ZONE_ID}/purge_cache",
#             headers={"Authorization": f"Bearer {CF_TOKEN}"},
#             json={"files": [f"https://mysite.com/blog/{event['slug']}"]}
#         )
#
#     message.ack()
#
# subscriber.subscribe(subscription_path, callback=callback)

from dataclasses import dataclass, field
from typing import List, Dict
from enum import Enum

class EventType(Enum):
    POST_CREATED = "post.created"
    POST_UPDATED = "post.updated"
    POST_DELETED = "post.deleted"
    MEDIA_UPLOADED = "media.uploaded"
    CACHE_PURGED = "cache.purged"

@dataclass
class PubSubEvent:
    event_type: EventType
    topic: str
    data: Dict
    timestamp: str

@dataclass
class Subscriber:
    name: str
    subscription: str
    action: str
    events: List[EventType]

subscribers = [
    Subscriber("Frontend Revalidator", "frontend-revalidate",
               "Revalidate Next.js ISR pages",
               [EventType.POST_CREATED, EventType.POST_UPDATED]),
    Subscriber("CDN Purger", "cdn-purge",
               "Purge Cloudflare cache for updated URLs",
               [EventType.POST_UPDATED, EventType.POST_DELETED]),
    Subscriber("Search Indexer", "search-index",
               "Update Algolia/Meilisearch index",
               [EventType.POST_CREATED, EventType.POST_UPDATED, EventType.POST_DELETED]),
    Subscriber("Newsletter", "newsletter-trigger",
               "Send newsletter for new posts",
               [EventType.POST_CREATED]),
]

print("\n=== Pub/Sub Subscribers ===")
for sub in subscribers:
    events = [e.value for e in sub.events]
    print(f"  [{sub.name}]")
    print(f"    Action: {sub.action}")
    print(f"    Events: {', '.join(events)}")

Architecture Diagram

# === Full Architecture ===

architecture_flow = [
    "1. Content Editor สร้าง/แก้ไข Post ใน WordPress Admin",
    "2. WordPress save_post Hook ส่ง Event ไป Pub/Sub Topic",
    "3. Pub/Sub กระจาย Event ไปทุก Subscriber",
    "4. Frontend Subscriber เรียก Next.js On-demand Revalidation",
    "5. CDN Subscriber เรียก Cloudflare Purge Cache",
    "6. Search Subscriber อัปเดต Search Index (Algolia)",
    "7. Newsletter Subscriber ส่งอีเมลแจ้ง Subscribers",
    "8. ผู้ใช้เห็น Content ใหม่ทันทีผ่าน CDN Edge",
]

print("=== Architecture Flow ===")
for step in architecture_flow:
    print(f"  {step}")

# Benefits
benefits = {
    "Decoupled": "WordPress ไม่ต้องรู้จัก Frontend CDN Search",
    "Scalable": "เพิ่ม Subscriber ใหม่ได้ไม่กระทบระบบเดิม",
    "Reliable": "Pub/Sub มี At-least-once Delivery Retry",
    "Real-time": "Content อัปเดตทันทีทุกช่องทาง",
    "Secure": "WordPress ไม่เปิด Public มี API Only",
}

print(f"\n\nBenefits:")
for benefit, desc in benefits.items():
    print(f"  [{benefit}]: {desc}")

# REST vs GraphQL
comparison = {
    "REST API": {
        "pros": "Simple, Built-in, No Plugin needed",
        "cons": "Over-fetching, Multiple requests",
        "use": "Simple projects, Mobile apps",
    },
    "WPGraphQL": {
        "pros": "Flexible queries, Single request, Type-safe",
        "cons": "Need Plugin, Learning curve",
        "use": "Complex frontends, Next.js, Gatsby",
    },
}

print(f"\n\nREST vs GraphQL:")
for api, info in comparison.items():
    print(f"\n  [{api}]")
    for k, v in info.items():
        print(f"    {k}: {v}")

เคล็ดลับ

• ISR: ใช้ Incremental Static Regeneration ใน Next.js สำหรับ Performance
• Preview: ตั้ง Preview Mode ให้ Editor ดู Draft ก่อน Publish
• Security: ปิด wp-login.php สำหรับ Public ใช้ IP Whitelist
• Webhook Secret: ใช้ Secret ยืนยันว่า Event มาจาก WordPress จริง
• Dead Letter: ตั้ง Dead Letter Queue สำหรับ Failed Messages

Headless WordPress คืออะไร

WordPress Backend CMS ไม่ใช้ Theme Frontend Next.js React REST API WPGraphQL Performance Security ดีกว่า

สรุป

WordPress Headless CMS REST API WPGraphQL Next.js Pub/Sub Event-driven Architecture ISR Revalidation CDN Purge Search Index Newsletter Decouple Scale Real-time Content Delivery