Technology

Directus CMS Progressive Delivery จัดการ Content ด้วย Feature Flags และ A/B Testing

directus cms progressive delivery
Directus CMS Progressive Delivery | SiamCafe Blog
2026-04-10· อ. บอม — SiamCafe.net· 1,012 คำ

Directus CMS ?????????????????????

Directus ???????????? open source headless CMS ???????????????????????? REST ????????? GraphQL API ????????? database ?????????????????? ?????????????????? SQL databases ??????????????????????????? (PostgreSQL, MySQL, SQLite, MS SQL, MariaDB) ???????????????????????????????????? API ????????? ???????????????????????? schema ?????? database ???????????? Directus ??????????????? API ????????????????????????????????????

Progressive Delivery ??????????????????????????????????????? deploy features ???????????????????????? ????????????????????????????????????????????????????????????????????? ???????????????????????????????????? ????????? feature flags, canary releases, A/B testing ?????????????????????????????????????????????????????? deploy ????????????????????? Progressive Delivery ???????????????????????? CMS ???????????????????????????????????????????????? content ????????????????????? users ???????????????????????????????????? ??????????????????????????????????????? rollout ????????????????????????

Directus ???????????????????????? Progressive Delivery ??????????????? API-first architecture ???????????????????????????????????? feature flag services ?????????????????????, Roles & Permissions ?????????????????? content access ?????????????????????, Webhooks ?????????????????????????????????????????? content ?????????????????????, Flows (automation) ??????????????? workflows ???????????????????????????, Versioning track content changes

????????????????????? Directus ?????????????????? Production

Setup Directus CMS

# === Directus Production Setup ===

# 1. Docker Compose Deployment
cat > docker-compose.yml << 'EOF'
version: '3.8'
services:
  directus:
    image: directus/directus:10.12
    ports:
      - "8055:8055"
    volumes:
      - directus_uploads:/directus/uploads
      - directus_extensions:/directus/extensions
    depends_on:
      - postgres
      - redis
    environment:
      KEY: ""
      SECRET: ""
      DB_CLIENT: "pg"
      DB_HOST: "postgres"
      DB_PORT: "5432"
      DB_DATABASE: "directus"
      DB_USER: "directus"
      DB_PASSWORD: ""
      CACHE_ENABLED: "true"
      CACHE_STORE: "redis"
      CACHE_REDIS: "redis://redis:6379"
      CACHE_AUTO_PURGE: "true"
      CACHE_TTL: "5m"
      ADMIN_EMAIL: "admin@example.com"
      ADMIN_PASSWORD: ""
      RATE_LIMITER_ENABLED: "true"
      RATE_LIMITER_STORE: "redis"
      CORS_ENABLED: "true"
      CORS_ORIGIN: "https://example.com"

  postgres:
    image: postgres:16-alpine
    volumes:
      - postgres_data:/var/lib/postgresql/data
    environment:
      POSTGRES_DB: directus
      POSTGRES_USER: directus
      POSTGRES_PASSWORD: ""

  redis:
    image: redis:7-alpine
    volumes:
      - redis_data:/data

volumes:
  directus_uploads:
  directus_extensions:
  postgres_data:
  redis_data:
EOF

# 2. Environment File
cat > .env << 'EOF'
DIRECTUS_KEY=random-key-here-change-this
DIRECTUS_SECRET=random-secret-here-change-this
DB_PASSWORD=strong-password-here
ADMIN_PASSWORD=admin-password-here
EOF

# 3. Start
docker compose up -d

# 4. Nginx Reverse Proxy
cat > /etc/nginx/sites-available/directus << 'EOF'
server {
    listen 443 ssl http2;
    server_name cms.example.com;

    ssl_certificate /etc/letsencrypt/live/cms.example.com/fullchain.pem;
    ssl_certificate_key /etc/letsencrypt/live/cms.example.com/privkey.pem;

    client_max_body_size 100m;

    location / {
        proxy_pass http://127.0.0.1:8055;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }
}
EOF

ln -s /etc/nginx/sites-available/directus /etc/nginx/sites-enabled/
nginx -t && systemctl reload nginx

echo "Directus production setup complete"

Progressive Delivery ????????? CMS

Implement progressive delivery ?????????????????? content

#!/usr/bin/env python3
# progressive_delivery.py ??? Progressive Delivery for CMS
import json
import logging
import hashlib
from typing import Dict, List, Optional
from datetime import datetime

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("delivery")

class ProgressiveDeliveryEngine:
    def __init__(self):
        self.feature_flags = {}
        self.experiments = {}
    
    def create_feature_flag(self, flag_id, name, default=False, rollout_pct=0):
        """Create a feature flag for content delivery"""
        self.feature_flags[flag_id] = {
            "id": flag_id,
            "name": name,
            "enabled": default,
            "rollout_percentage": rollout_pct,
            "targeting_rules": [],
            "created_at": datetime.utcnow().isoformat(),
        }
        return self.feature_flags[flag_id]
    
    def add_targeting_rule(self, flag_id, rule):
        """Add targeting rule to a feature flag"""
        if flag_id in self.feature_flags:
            self.feature_flags[flag_id]["targeting_rules"].append(rule)
    
    def evaluate_flag(self, flag_id, user_context):
        """Evaluate if a user should see the feature"""
        flag = self.feature_flags.get(flag_id)
        if not flag or not flag["enabled"]:
            return False
        
        # Check targeting rules
        for rule in flag["targeting_rules"]:
            if rule["type"] == "user_segment":
                if user_context.get("segment") in rule["segments"]:
                    return True
            elif rule["type"] == "country":
                if user_context.get("country") in rule["countries"]:
                    return True
        
        # Percentage-based rollout
        if flag["rollout_percentage"] > 0:
            user_hash = int(hashlib.md5(
                f"{flag_id}:{user_context.get('user_id', '')}".encode()
            ).hexdigest(), 16) % 100
            return user_hash < flag["rollout_percentage"]
        
        return False
    
    def get_content_variant(self, content_id, user_context):
        """Get the right content variant for a user"""
        flag_id = f"content_{content_id}"
        
        if self.evaluate_flag(flag_id, user_context):
            return {"variant": "B", "content_id": f"{content_id}_v2"}
        return {"variant": "A", "content_id": content_id}
    
    def create_canary_release(self, content_id, stages):
        """Create a canary release plan"""
        return {
            "content_id": content_id,
            "stages": stages,
            "current_stage": 0,
            "status": "active",
            "example_stages": [
                {"name": "Internal", "percentage": 1, "duration": "1 hour"},
                {"name": "Canary", "percentage": 5, "duration": "4 hours"},
                {"name": "Early Access", "percentage": 25, "duration": "24 hours"},
                {"name": "General", "percentage": 100, "duration": "permanent"},
            ],
        }

engine = ProgressiveDeliveryEngine()

flag = engine.create_feature_flag("content_homepage_v2", "New Homepage Content", default=True, rollout_pct=25)
engine.add_targeting_rule("content_homepage_v2", {"type": "user_segment", "segments": ["beta_testers"]})

users = [
    {"user_id": "user_1", "segment": "beta_testers", "country": "TH"},
    {"user_id": "user_2", "segment": "regular", "country": "TH"},
    {"user_id": "user_3", "segment": "regular", "country": "US"},
]

for user in users:
    result = engine.evaluate_flag("content_homepage_v2", user)
    print(f"  {user['user_id']} ({user['segment']}): {'New content' if result else 'Original'}")

canary = engine.create_canary_release("homepage", [])
print(f"\nCanary stages: {len(canary['example_stages'])}")

Feature Flags ????????? A/B Testing

A/B Testing ?????????????????? CMS content

#!/usr/bin/env python3
# ab_testing.py ??? CMS A/B Testing Framework
import json
import logging
import hashlib
import random
from typing import Dict, List

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("abtest")

class CMSABTesting:
    def __init__(self):
        self.experiments = {}
        self.results = {}
    
    def create_experiment(self, exp_id, name, variants, traffic_pct=100):
        """Create an A/B test experiment"""
        self.experiments[exp_id] = {
            "id": exp_id,
            "name": name,
            "variants": variants,
            "traffic_percentage": traffic_pct,
            "status": "running",
            "results": {v["id"]: {"views": 0, "clicks": 0, "conversions": 0} for v in variants},
            "created_at": "2024-01-01",
        }
        return self.experiments[exp_id]
    
    def assign_variant(self, exp_id, user_id):
        """Assign a user to an experiment variant"""
        exp = self.experiments.get(exp_id)
        if not exp or exp["status"] != "running":
            return exp["variants"][0] if exp else None
        
        # Deterministic assignment based on user_id
        hash_val = int(hashlib.md5(f"{exp_id}:{user_id}".encode()).hexdigest(), 16)
        
        # Check if user is in traffic allocation
        if (hash_val % 100) >= exp["traffic_percentage"]:
            return None  # Not in experiment
        
        # Assign to variant
        variant_idx = hash_val % len(exp["variants"])
        variant = exp["variants"][variant_idx]
        
        # Track view
        exp["results"][variant["id"]]["views"] += 1
        
        return variant
    
    def track_conversion(self, exp_id, variant_id):
        """Track a conversion event"""
        exp = self.experiments.get(exp_id)
        if exp and variant_id in exp["results"]:
            exp["results"][variant_id]["conversions"] += 1
    
    def get_results(self, exp_id):
        """Get experiment results with statistical analysis"""
        exp = self.experiments.get(exp_id)
        if not exp:
            return None
        
        results = []
        for variant in exp["variants"]:
            vid = variant["id"]
            data = exp["results"][vid]
            cvr = data["conversions"] / max(data["views"], 1) * 100
            
            results.append({
                "variant": vid,
                "content": variant.get("content_id", ""),
                "views": data["views"],
                "conversions": data["conversions"],
                "conversion_rate": round(cvr, 2),
            })
        
        # Determine winner
        if len(results) >= 2:
            sorted_results = sorted(results, key=lambda x: x["conversion_rate"], reverse=True)
            winner = sorted_results[0]
            lift = ((winner["conversion_rate"] - sorted_results[1]["conversion_rate"]) /
                    max(sorted_results[1]["conversion_rate"], 0.01)) * 100
        else:
            winner = results[0] if results else None
            lift = 0
        
        return {
            "experiment": exp["name"],
            "status": exp["status"],
            "variants": results,
            "winner": winner["variant"] if winner else None,
            "lift": round(lift, 1),
        }

ab = CMSABTesting()

ab.create_experiment("hero_banner", "Homepage Hero Banner Test", [
    {"id": "control", "content_id": "banner_original", "description": "Original banner"},
    {"id": "variant_b", "content_id": "banner_new", "description": "New design with CTA"},
])

# Simulate traffic
for i in range(1000):
    variant = ab.assign_variant("hero_banner", f"user_{i}")
    if variant and random.random() < (0.03 if variant["id"] == "control" else 0.045):
        ab.track_conversion("hero_banner", variant["id"])

results = ab.get_results("hero_banner")
print(f"Experiment: {results['experiment']}")
for v in results["variants"]:
    print(f"  {v['variant']}: {v['views']} views, {v['conversions']} conv, {v['conversion_rate']}% CVR")
print(f"Winner: {results['winner']} (lift: {results['lift']}%)")

API-First Content Delivery

????????? Directus API ?????????????????? content delivery

# === Directus API Content Delivery ===

# 1. REST API Examples

# Get all published articles
curl -s "https://cms.example.com/items/articles?filter[status][_eq]=published&sort=-date_created&limit=10" \
  -H "Authorization: Bearer YOUR_TOKEN" | jq .

# Get single article with related data
curl -s "https://cms.example.com/items/articles/1?fields=*, author.name, category.title, tags.*" \
  -H "Authorization: Bearer YOUR_TOKEN" | jq .

# 2. GraphQL Query
cat > query.graphql << 'EOF'
query GetArticles($status: String!, $limit: Int!) {
  articles(
    filter: { status: { _eq: $status } }
    sort: ["-date_created"]
    limit: $limit
  ) {
    id
    title
    slug
    excerpt
    content
    featured_image {
      id
      filename_download
    }
    author {
      first_name
      last_name
    }
    category {
      title
      slug
    }
    date_created
    date_updated
  }
}
EOF

# 3. SDK Integration (JavaScript)
cat > directus-client.js << 'JSEOF'
import { createDirectus, rest, readItems, readItem } from '@directus/sdk';

const client = createDirectus('https://cms.example.com').with(rest());

// Get published articles
async function getArticles(page = 1, limit = 10) {
  return await client.request(
    readItems('articles', {
      filter: { status: { _eq: 'published' } },
      sort: ['-date_created'],
      limit,
      page,
      fields: ['id', 'title', 'slug', 'excerpt', 'featured_image', 'date_created'],
    })
  );
}

// Get article with feature flag check
async function getArticleWithFlags(slug, userId) {
  const article = await client.request(
    readItem('articles', slug, {
      fields: ['*', 'author.*', 'category.*'],
    })
  );
  
  // Check for A/B test variant
  const variant = await checkFeatureFlag('article_layout_v2', userId);
  
  return {
    ...article,
    layout: variant ? 'new_layout' : 'default_layout',
  };
}

console.log('Directus client configured');
JSEOF

# 4. Webhook Configuration for CDN Purge
cat > webhook_handler.py << 'PYEOF'
#!/usr/bin/env python3
from fastapi import FastAPI, Request
import httpx

app = FastAPI()

@app.post("/webhook/content-updated")
async def content_updated(request: Request):
    payload = await request.json()
    collection = payload.get("collection")
    keys = payload.get("keys", [])
    
    # Purge CDN cache for updated content
    for key in keys:
        cache_key = f"/{collection}/{key}"
        # Cloudflare purge
        async with httpx.AsyncClient() as client:
            await client.post(
                "https://api.cloudflare.com/client/v4/zones/ZONE_ID/purge_cache",
                json={"files": [f"https://example.com{cache_key}"]},
                headers={"Authorization": "Bearer CF_TOKEN"},
            )
    
    return {"status": "purged", "keys": keys}
PYEOF

echo "API content delivery configured"

Monitoring ????????? Rollback

Monitor content delivery ????????? rollback

#!/usr/bin/env python3
# content_monitor.py ??? Content Delivery Monitoring
import json
import logging
from typing import Dict

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("monitor")

class ContentDeliveryMonitor:
    def __init__(self):
        self.metrics = {}
    
    def dashboard(self):
        return {
            "content_delivery": {
                "total_api_requests_today": 125000,
                "cache_hit_rate": 94.5,
                "avg_response_time_ms": 45,
                "p95_response_time_ms": 120,
                "error_rate": 0.1,
            },
            "progressive_delivery": {
                "active_experiments": 3,
                "active_feature_flags": 12,
                "canary_releases": 1,
                "rollbacks_today": 0,
            },
            "content_stats": {
                "total_items": 2500,
                "published": 2100,
                "draft": 350,
                "archived": 50,
                "updated_today": 15,
            },
        }
    
    def rollback_strategy(self):
        return {
            "automatic_rollback": {
                "triggers": [
                    "Error rate > 5% for 5 minutes",
                    "Response time > 500ms (p95) for 10 minutes",
                    "Conversion rate drops > 20% compared to baseline",
                ],
                "action": "Disable feature flag, revert to original content",
            },
            "manual_rollback": {
                "steps": [
                    "1. Identify issue in monitoring dashboard",
                    "2. Disable feature flag / stop experiment",
                    "3. Purge CDN cache",
                    "4. Verify rollback successful",
                    "5. Investigate root cause",
                    "6. Fix and re-deploy with smaller rollout",
                ],
            },
            "content_versioning": {
                "directus_revisions": "Directus tracks all content changes automatically",
                "rollback_content": "Revert to previous revision via API or Admin UI",
                "api_endpoint": "POST /revisions/:id/revert",
            },
        }

monitor = ContentDeliveryMonitor()
dash = monitor.dashboard()
print(f"API Requests: {dash['content_delivery']['total_api_requests_today']:,}")
print(f"Cache Hit: {dash['content_delivery']['cache_hit_rate']}%")
print(f"Active Experiments: {dash['progressive_delivery']['active_experiments']}")

strategy = monitor.rollback_strategy()
print(f"\nAuto-rollback triggers: {len(strategy['automatic_rollback']['triggers'])}")
for trigger in strategy["automatic_rollback"]["triggers"]:
    print(f"  - {trigger}")

FAQ ??????????????????????????????????????????

Q: Directus ????????? Strapi ???????????????????????????????????????????

A: Directus ??????????????????????????? existing database ?????????????????? ???????????????????????? schema ???????????? ?????????????????? SQL databases ??????????????????????????? ?????? REST + GraphQL API ??????????????????????????? Data Studio UI ????????? ?????????????????????????????? ??????????????????????????? TypeScript + Vue.js Strapi ??????????????? schema ???????????? Content-Type Builder ????????? own database schema ?????????????????? PostgreSQL, MySQL, SQLite ?????? REST + GraphQL ???????????? plugin Community ???????????????????????? Marketplace ?????? plugins ???????????? ??????????????????????????? JavaScript + React ??????????????? Directus ????????????????????? existing database ????????????????????? flexibility ????????? ??????????????? Strapi ??????????????????????????????????????????????????? ????????????????????? ecosystem ?????????????????????

Q: Progressive Delivery ????????????????????? Continuous Delivery ??????????????????????

A: Continuous Delivery deploy ????????? change ?????? production ??????????????????????????? ?????????????????????????????????????????????????????? ??????????????? bug ?????????????????????????????????????????????????????? Progressive Delivery deploy change ???????????????????????? ??????????????????????????????????????????????????? (1%) ????????????????????????????????? (5%, 25%, 100%) ????????? feature flags ?????????????????? ???????????????????????????????????? rollback ????????????????????????????????? ?????????????????? Continuous Delivery + safety net ?????????????????? CMS content Progressive Delivery ??????????????????????????? content ????????????????????? users ???????????????????????? ??????????????? (engagement, conversion) ???????????????????????? rollout ????????????????????????

Q: Feature Flags ?????????????????????????????????????????????????

A: ?????? Technical debt ?????????????????? cleanup flags ??????????????????????????????????????? code ????????????????????????????????????????????????????????????, Testing complexity ???????????? test ????????? combination ????????? flags, Performance overhead evaluate flags ????????? request (????????????????????? caching), Management ?????????????????? process ?????????????????? flags lifecycle ?????????????????????????????? ???????????? expiration date ??????????????????????????? flag, Review flags ????????? sprint ?????? flags ???????????????????????????, ????????? feature flag service (LaunchDarkly, Unleash, Flagsmith) ????????????????????????, Document ????????? flag ???????????????????????????????????? ??????????????????????????????????????????

Q: Directus ?????????????????? multi-language content ??????????

A: ?????????????????? Directus ?????? built-in translations interface ??????????????? translations collection ??????????????????????????? content collection ???????????? ??????????????? languages ??????????????????????????? (TH, EN, JA, etc.) ??????????????? content item ?????? translations ???????????????????????? API query ???????????? ?deep[translations][_filter][languages_code][_eq]=th ????????? content ?????????????????????????????????????????? ?????????????????? Progressive Delivery ??????????????? translation ????????????????????? users ??????????????????????????????????????? ???????????? ???????????????????????? rollout

📖 บทความที่เกี่ยวข้อง

Distributed Tracing Progressive Deliveryอ่านบทความ → C# Blazor Progressive Deliveryอ่านบทความ → WordPress WooCommerce Progressive Deliveryอ่านบทความ → Azure Front Door Progressive Deliveryอ่านบทความ → Directus CMS Metric Collectionอ่านบทความ →

📚 ดูบทความทั้งหมด →