Supabase Realtime Feature Flag Management

Supabase Feature Flag

Supabase Realtime Feature Flag PostgreSQL WebSocket RLS Gradual Rollout A/B Testing Kill Switch Toggle User Targeting

Database Schema

# === Supabase Feature Flag Schema ===

# -- Feature Flags Table
# CREATE TABLE feature_flags (
#     id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
#     name TEXT UNIQUE NOT NULL,
#     description TEXT,
#     enabled BOOLEAN DEFAULT false,
#     percentage INTEGER DEFAULT 100 CHECK(percentage BETWEEN 0 AND 100),
#     target_rules JSONB DEFAULT '{}',
#     flag_type TEXT DEFAULT 'boolean' CHECK(flag_type IN ('boolean','percentage','targeted','ab_test')),
#     created_at TIMESTAMPTZ DEFAULT now(),
#     updated_at TIMESTAMPTZ DEFAULT now(),
#     created_by UUID REFERENCES auth.users(id)
# );
#
# -- Flag Overrides (per-user)
# CREATE TABLE flag_overrides (
#     id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
#     flag_id UUID REFERENCES feature_flags(id) ON DELETE CASCADE,
#     user_id UUID REFERENCES auth.users(id),
#     enabled BOOLEAN NOT NULL,
#     reason TEXT,
#     created_at TIMESTAMPTZ DEFAULT now(),
#     UNIQUE(flag_id, user_id)
# );
#
# -- Audit Log
# CREATE TABLE flag_events (
#     id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
#     flag_id UUID REFERENCES feature_flags(id),
#     action TEXT NOT NULL,
#     old_value JSONB,
#     new_value JSONB,
#     changed_by UUID REFERENCES auth.users(id),
#     changed_at TIMESTAMPTZ DEFAULT now()
# );
#
# -- RLS Policies
# ALTER TABLE feature_flags ENABLE ROW LEVEL SECURITY;
# CREATE POLICY "Anyone can read flags" ON feature_flags FOR SELECT USING (true);
# CREATE POLICY "Admins can modify flags" ON feature_flags FOR ALL
#   USING (auth.jwt() ->> 'role' = 'admin');
#
# -- Enable Realtime
# ALTER PUBLICATION supabase_realtime ADD TABLE feature_flags;

from dataclasses import dataclass, field

@dataclass
class FeatureFlag:
    name: str
    flag_type: str
    enabled: bool
    percentage: int
    target_rules: dict
    description: str

flags = [
    FeatureFlag("new_checkout_flow", "percentage", True, 25,
        {"exclude_countries": ["CN"]},
        "New checkout UI - gradual rollout 25%"),
    FeatureFlag("dark_mode", "boolean", True, 100,
        {},
        "Dark mode toggle - fully rolled out"),
    FeatureFlag("ai_recommendations", "targeted", True, 100,
        {"plans": ["premium", "enterprise"]},
        "AI product recommendations - premium only"),
    FeatureFlag("button_color_test", "ab_test", True, 100,
        {"variants": {"A": "blue", "B": "green"}, "split": 50},
        "A/B test button color"),
    FeatureFlag("legacy_api", "boolean", False, 0,
        {},
        "Kill switch for legacy API - disabled"),
]

print("=== Feature Flags ===")
for f in flags:
    status = "ON" if f.enabled else "OFF"
    print(f"  [{f.name}] {f.flag_type} [{status}] {f.percentage}%")
    print(f"    Rules: {f.target_rules}")
    print(f"    Desc: {f.description}")

Realtime Client

# === Supabase Realtime Feature Flag Client ===

# // JavaScript Client
# import { createClient } from '@supabase/supabase-js';
#
# const supabase = createClient(SUPABASE_URL, SUPABASE_ANON_KEY);
#
# // Cache flags locally
# let flagCache = {};
#
# // Load initial flags
# async function loadFlags() {
#   const { data } = await supabase
#     .from('feature_flags')
#     .select('*')
#     .eq('enabled', true);
#   data.forEach(flag => flagCache[flag.name] = flag);
# }
#
# // Subscribe to realtime changes
# supabase
#   .channel('feature-flags')
#   .on('postgres_changes',
#     { event: '*', schema: 'public', table: 'feature_flags' },
#     (payload) => {
#       if (payload.eventType === 'UPDATE' || payload.eventType === 'INSERT') {
#         flagCache[payload.new.name] = payload.new;
#       } else if (payload.eventType === 'DELETE') {
#         delete flagCache[payload.old.name];
#       }
#       console.log('Flag updated:', payload.new?.name);
#     }
#   )
#   .subscribe();
#
# // Check flag
# function isEnabled(flagName, userId, userContext) {
#   const flag = flagCache[flagName];
#   if (!flag || !flag.enabled) return false;
#   if (flag.flag_type === 'boolean') return true;
#   if (flag.flag_type === 'percentage') {
#     return hashUser(userId) % 100 < flag.percentage;
#   }
#   if (flag.flag_type === 'targeted') {
#     return matchRules(flag.target_rules, userContext);
#   }
#   return false;
# }

@dataclass
class RealtimeEvent:
    event_type: str
    flag_name: str
    old_value: str
    new_value: str
    latency: str

events = [
    RealtimeEvent("UPDATE", "new_checkout_flow",
        "percentage: 25", "percentage: 50",
        "< 500ms to all clients"),
    RealtimeEvent("UPDATE", "ai_recommendations",
        "plans: [premium]", "plans: [premium, pro]",
        "< 500ms"),
    RealtimeEvent("UPDATE", "legacy_api",
        "enabled: true", "enabled: false (Kill Switch!)",
        "< 200ms (urgent)"),
    RealtimeEvent("INSERT", "holiday_banner",
        "-", "enabled: true, percentage: 100",
        "< 500ms"),
]

print("=== Realtime Events ===")
for e in events:
    print(f"  [{e.event_type}] {e.flag_name}")
    print(f"    Old: {e.old_value} → New: {e.new_value}")
    print(f"    Latency: {e.latency}")

A/B Testing & Analytics

# === A/B Testing with Feature Flags ===

@dataclass
class ABTest:
    test_name: str
    variants: dict
    metric: str
    duration: str
    result: str

tests = [
    ABTest("button_color_test",
        {"A": "Blue Button (Control)", "B": "Green Button (Variant)"},
        "Click-through Rate (CTR)",
        "2 สัปดาห์",
        "B (Green): CTR 4.2% vs A (Blue): 3.1% → B wins (+35%)"),
    ABTest("checkout_steps",
        {"A": "3-step Checkout", "B": "1-page Checkout"},
        "Conversion Rate",
        "4 สัปดาห์",
        "B (1-page): 5.8% vs A (3-step): 4.5% → B wins (+29%)"),
    ABTest("pricing_display",
        {"A": "Monthly Price", "B": "Annual Price (savings shown)"},
        "Signup Rate",
        "3 สัปดาห์",
        "B (Annual): 7.1% vs A (Monthly): 6.2% → B wins (+15%)"),
]

print("=== A/B Tests ===")
for t in tests:
    print(f"\n  [{t.test_name}] Duration: {t.duration}")
    for k, v in t.variants.items():
        print(f"    Variant {k}: {v}")
    print(f"    Metric: {t.metric}")
    print(f"    Result: {t.result}")

เคล็ดลับ

• Cache: Cache Flag ที่ Client ลด Latency ใช้ Realtime อัพเดท
• RLS: ใช้ Row Level Security ป้องกัน Unauthorized Flag Change
• Cleanup: ลบ Flag เก่าที่ Rollout 100% แล้ว ลด Technical Debt
• Audit: บันทึกทุกการเปลี่ยนแปลง Flag ใน flag_events
• Kill Switch: ทุก External Dependency ต้องมี Kill Switch Flag

Supabase คืออะไร

Open Source BaaS PostgreSQL Realtime WebSocket Authentication Storage Edge Functions RLS Firebase Alternative JavaScript Python Dart

สรุป

Supabase Realtime Feature Flag PostgreSQL WebSocket RLS Gradual Rollout A/B Testing Kill Switch Cache Audit Targeting Production