【系统架构设计百科】特性开关架构：安全发布与实验的工程实践

if (featureFlags.isEnabled("new-checkout-flow")) {
    return newCheckoutService.process(order);
} else {
    return legacyCheckoutService.process(order);
}

# flag 配置
new-search-algorithm:
  type: release
  owner: search-team
  created: 2026-03-01
  expected-removal: 2026-04-01
  description: "新的搜索排序算法，目前仅对内部用户开放"
  default: false
  targeting:
    - segment: internal-employees
      value: true

# 实验开关示例：按钮颜色 A/B 测试
variant = feature_flags.get_variant(
    "checkout-button-color",
    user_id=user.id,
    default="blue"
)

if variant == "blue":
    render_blue_button()
elif variant == "green":
    render_green_button()
elif variant == "orange":
    render_orange_button()

# 上报实验事件
analytics.track("checkout-button-impression", {
    "variant": variant,
    "user_id": user.id,
    "session_id": session.id
})

// 运维开关示例：推荐服务降级
func GetRecommendations(ctx context.Context, userID string) ([]Product, error) {
    if !featureFlags.IsEnabled("recommendation-service-enabled") {
        // 降级到热门商品列表
        return getPopularProducts(ctx)
    }

    resp, err := recommendationClient.GetPersonalized(ctx, userID)
    if err != nil {
        return getPopularProducts(ctx)
    }
    return resp.Products, nil
}

// 权限开关示例：高级功能控制
async function renderDashboard(user: User): Promise<Dashboard> {
    const dashboard = new Dashboard();

    dashboard.addWidget(new BasicAnalytics());

    if (await featureFlags.isEnabled("advanced-analytics", { userId: user.id })) {
        dashboard.addWidget(new AdvancedAnalytics());
        dashboard.addWidget(new PredictiveInsights());
    }

    if (await featureFlags.isEnabled("custom-reports", { userId: user.id })) {
        dashboard.addWidget(new CustomReportBuilder());
    }

    return dashboard;
}

# 轮询模式的伪代码
class PollingFlagClient:
    def __init__(self, api_url: str, poll_interval: int = 30):
        self._cache: dict[str, FlagDefinition] = {}
        self._api_url = api_url
        self._poll_interval = poll_interval
        self._start_polling()

    def _start_polling(self):
        """后台线程定期拉取 flag 定义"""
        def poll_loop():
            while True:
                try:
                    response = requests.get(
                        f"{self._api_url}/flags",
                        headers={"If-None-Match": self._etag}
                    )
                    if response.status_code == 200:
                        self._cache = response.json()
                        self._etag = response.headers.get("ETag")
                except Exception as e:
                    logger.warning(f"Flag polling failed: {e}")
                time.sleep(self._poll_interval)

        thread = threading.Thread(target=poll_loop, daemon=True)
        thread.start()

    def is_enabled(self, flag_key: str, context: dict = None) -> bool:
        flag_def = self._cache.get(flag_key)
        if flag_def is None:
            return False  # 未知 flag 返回 false（安全默认值）
        return self._evaluate(flag_def, context)

// 流式推送模式的伪代码（Java）
public class StreamingFlagClient implements FlagClient {
    private final ConcurrentHashMap<String, FlagDefinition> cache;
    private final EventSource eventSource;

    public StreamingFlagClient(String streamUrl, String sdkKey) {
        this.cache = new ConcurrentHashMap<>();

        EventHandler handler = new EventHandler() {
            @Override
            public void onMessage(String event, MessageEvent messageEvent) {
                if ("put".equals(event)) {
                    // 全量更新
                    Map<String, FlagDefinition> allFlags =
                        JsonParser.parseFlags(messageEvent.getData());
                    cache.clear();
                    cache.putAll(allFlags);
                } else if ("patch".equals(event)) {
                    // 增量更新单个 flag
                    FlagPatch patch = JsonParser.parsePatch(messageEvent.getData());
                    cache.put(patch.getKey(), patch.getDefinition());
                } else if ("delete".equals(event)) {
                    String key = JsonParser.parseDeleteKey(messageEvent.getData());
                    cache.remove(key);
                }
            }

            @Override
            public void onError(Throwable t) {
                logger.error("SSE connection error, will retry", t);
            }
        };

        this.eventSource = new EventSource.Builder(handler, URI.create(streamUrl))
            .header("Authorization", sdkKey)
            .reconnectTime(Duration.ofSeconds(1))
            .build();
        this.eventSource.start();
    }

    @Override
    public boolean isEnabled(String flagKey, EvaluationContext context) {
        FlagDefinition def = cache.get(flagKey);
        if (def == null) {
            return false;
        }
        return evaluate(def, context);
    }
}

{
  "key": "new-checkout-flow",
  "version": 42,
  "on": true,
  "prerequisites": [
    { "key": "payment-service-v2", "variation": 0 }
  ],
  "targets": [
    {
      "variation": 0,
      "values": ["user-123", "user-456"]
    }
  ],
  "rules": [
    {
      "id": "rule-1",
      "clauses": [
        {
          "attribute": "country",
          "op": "in",
          "values": ["CN", "JP", "KR"]
        },
        {
          "attribute": "plan",
          "op": "in",
          "values": ["enterprise", "pro"]
        }
      ],
      "variation_or_rollout": {
        "rollout": {
          "variations": [
            { "variation": 0, "weight": 80000 },
            { "variation": 1, "weight": 20000 }
          ],
          "bucket_by": "user_id"
        }
      }
    }
  ],
  "fallthrough": {
    "variation": 1
  },
  "off_variation": 1,
  "variations": [true, false],
  "salt": "abc123"
}

// 一致性哈希分桶实现
func bucketUser(flagKey, salt, userID, bucketBy string) float64 {
    key := fmt.Sprintf("%s.%s.%s", flagKey, salt, userID)
    hash := sha1.Sum([]byte(key))
    // 取前 15 个十六进制字符，转换为整数
    hashStr := hex.EncodeToString(hash[:])[:15]
    intVal, _ := strconv.ParseInt(hashStr, 16, 64)
    // 归一化到 [0, 1) 区间
    return float64(intVal) / float64(0x1000000000000000)
}

func evaluateRollout(rollout Rollout, flagKey, salt string, context EvalContext) int {
    bucketVal := bucketUser(flagKey, salt, context.GetAttribute(rollout.BucketBy), "")
    var sum float64
    for _, wv := range rollout.Variations {
        sum += float64(wv.Weight) / 100000.0
        if bucketVal < sum {
            return wv.Variation
        }
    }
    return rollout.Variations[len(rollout.Variations)-1].Variation
}

// LaunchDarkly 服务端 SDK 使用示例（Java）
LDConfig config = new LDConfig.Builder()
    .events(
        Components.sendEvents()
            .capacity(10000)
            .flushInterval(Duration.ofSeconds(5))
    )
    .dataSource(
        Components.streamingDataSource()
            .initialReconnectDelay(Duration.ofSeconds(1))
    )
    .build();

LDClient client = new LDClient("sdk-key-server-side", config);

// 求值 flag——完全在本地执行，无网络调用
LDContext context = LDContext.builder("user-123")
    .set("plan", "enterprise")
    .set("country", "CN")
    .build();

boolean showNewFeature = client.boolVariation(
    "new-checkout-flow",
    context,
    false  // 默认值，当 flag 不存在或求值失败时使用
);

// LaunchDarkly 客户端 SDK 使用示例（JavaScript）
import * as LDClient from "launchdarkly-js-client-sdk";

const context = {
    kind: "user",
    key: "user-123",
    name: "Alice",
    email: "alice@example.com",
    custom: {
        plan: "enterprise",
        country: "CN"
    }
};

const client = LDClient.initialize("client-side-id", context);

client.on("ready", () => {
    const showNewFeature = client.variation("new-checkout-flow", false);
    if (showNewFeature) {
        renderNewCheckoutFlow();
    } else {
        renderLegacyCheckoutFlow();
    }
});

// 监听 flag 值变更
client.on("change:new-checkout-flow", (current, previous) => {
    console.log(`Flag changed from ${previous} to ${current}`);
    rerender();
});

# LaunchDarkly Relay Proxy 配置
main:
  port: 8030
  heartbeatInterval: 15s

environment:
  production:
    sdkKey: "sdk-key-production"
    mobileKey: "mob-key-production"
    envId: "client-side-id-production"
    prefix: "ld-prod"

  staging:
    sdkKey: "sdk-key-staging"

redis:
  host: "redis.internal"
  port: 6379
  localTtl: 30s

{
  "kind": "feature",
  "creationDate": 1712000000000,
  "key": "new-checkout-flow",
  "version": 42,
  "context": {
    "kind": "user",
    "key": "user-123"
  },
  "variation": 0,
  "value": true,
  "default": false,
  "reason": {
    "kind": "RULE_MATCH",
    "ruleIndex": 0,
    "ruleId": "rule-1"
  }
}

// Unleash 内置激活策略
interface ActivationStrategy {
    name: string;
    parameters: Record<string, string>;
    constraints?: Constraint[];
}

// 1. 标准策略（Standard）：对所有人开放
const standardStrategy: ActivationStrategy = {
    name: "default",
    parameters: {}
};

// 2. 渐进发布策略（Gradual Rollout）
const gradualRollout: ActivationStrategy = {
    name: "flexibleRollout",
    parameters: {
        rollout: "30",           // 30% 的用户
        stickiness: "userId",    // 按 userId 做一致性分桶
        groupId: "new-checkout"  // 分桶标识
    }
};

// 3. 用户 ID 策略（UserIDs）
const userIdStrategy: ActivationStrategy = {
    name: "userWithId",
    parameters: {
        userIds: "user-123,user-456,user-789"
    }
};

// 4. IP 地址策略（IPs）
const ipStrategy: ActivationStrategy = {
    name: "remoteAddress",
    parameters: {
        IPs: "10.0.0.0/8,192.168.1.0/24"
    }
};

// 5. 自定义策略
const customStrategy: ActivationStrategy = {
    name: "betaUsers",
    parameters: {
        minAccountAge: "90",
        requiredPlan: "pro,enterprise"
    },
    constraints: [
        {
            contextName: "environment",
            operator: "IN",
            values: ["production", "staging"]
        }
    ]
};

// Java SDK
UnleashConfig config = UnleashConfig.builder()
    .appName("my-service")
    .instanceId("instance-1")
    .unleashAPI("http://unleash.internal:4242/api")
    .apiKey("default:development.abc123")
    .synchronousFetchOnInitialisation(true)
    .fetchTogglesInterval(10)  // 每 10 秒轮询一次
    .sendMetricsInterval(60)   // 每 60 秒上报指标
    .build();

Unleash unleash = new DefaultUnleash(config);

// 简单布尔判断
if (unleash.isEnabled("new-checkout-flow")) {
    newCheckout();
}

// 带上下文的判断
UnleashContext context = UnleashContext.builder()
    .userId("user-123")
    .addProperty("plan", "enterprise")
    .addProperty("country", "CN")
    .build();

if (unleash.isEnabled("premium-feature", context)) {
    showPremiumFeature();
}

// 获取变体（用于 A/B 测试）
Variant variant = unleash.getVariant("checkout-button-color", context);
String buttonColor = variant.getPayload()
    .map(Payload::getValue)
    .orElse("blue");

# Python SDK
from UnleashClient import UnleashClient

client = UnleashClient(
    url="http://unleash.internal:4242/api",
    app_name="my-python-service",
    custom_headers={"Authorization": "default:development.abc123"}
)
client.initialize_client()

# 简单判断
if client.is_enabled("new-search-algorithm"):
    use_new_search()

# 带上下文
context = {
    "userId": "user-123",
    "properties": {
        "plan": "enterprise",
        "region": "asia-pacific"
    }
}

if client.is_enabled("regional-pricing", context):
    apply_regional_pricing()

# 变体
variant = client.get_variant("onboarding-flow", context)
if variant.get("name") == "simplified":
    show_simplified_onboarding()
elif variant.get("name") == "guided":
    show_guided_onboarding()

# Unleash Edge 配置
upstream:
  url: "http://unleash.internal:4242/api"
  tokens:
    - "default:production.xyz789"
  refresh_interval_seconds: 10

server:
  port: 3063
  host: "0.0.0.0"

features:
  offline_mode: false
  token_validation: strict

// 最小自建 flag 服务（Go）
package main

import (
    "encoding/json"
    "log"
    "net/http"
    "sync"
)

type Flag struct {
    Key          string            `json:"key"`
    Enabled      bool              `json:"enabled"`
    Percentage   float64           `json:"percentage,omitempty"`
    AllowList    []string          `json:"allow_list,omitempty"`
    Variants     map[string]int    `json:"variants,omitempty"`
    Metadata     map[string]string `json:"metadata,omitempty"`
}

type FlagStore struct {
    mu    sync.RWMutex
    flags map[string]*Flag
    // SSE 订阅者
    subscribers map[chan []byte]struct{}
    subMu       sync.RWMutex
}

func (s *FlagStore) GetAll() []*Flag {
    s.mu.RLock()
    defer s.mu.RUnlock()
    result := make([]*Flag, 0, len(s.flags))
    for _, f := range s.flags {
        result = append(result, f)
    }
    return result
}

func (s *FlagStore) Update(flag *Flag) {
    s.mu.Lock()
    s.flags[flag.Key] = flag
    s.mu.Unlock()
    // 通知所有 SSE 订阅者
    data, _ := json.Marshal(flag)
    s.subMu.RLock()
    for ch := range s.subscribers {
        select {
        case ch <- data:
        default:
            // 订阅者消费太慢，跳过
        }
    }
    s.subMu.RUnlock()
}

func (s *FlagStore) Subscribe() chan []byte {
    ch := make(chan []byte, 100)
    s.subMu.Lock()
    s.subscribers[ch] = struct{}{}
    s.subMu.Unlock()
    return ch
}

func (s *FlagStore) Unsubscribe(ch chan []byte) {
    s.subMu.Lock()
    delete(s.subscribers, ch)
    s.subMu.Unlock()
    close(ch)
}

// SSE 端点
func (s *FlagStore) StreamHandler(w http.ResponseWriter, r *http.Request) {
    flusher, ok := w.(http.Flusher)
    if !ok {
        http.Error(w, "streaming not supported", http.StatusInternalServerError)
        return
    }

    w.Header().Set("Content-Type", "text/event-stream")
    w.Header().Set("Cache-Control", "no-cache")
    w.Header().Set("Connection", "keep-alive")

    // 先发送当前所有 flag 的全量数据
    allFlags := s.GetAll()
    data, _ := json.Marshal(allFlags)
    w.Write([]byte("event: put\ndata: "))
    w.Write(data)
    w.Write([]byte("\n\n"))
    flusher.Flush()

    // 订阅增量更新
    ch := s.Subscribe()
    defer s.Unsubscribe(ch)

    for {
        select {
        case msg := <-ch:
            w.Write([]byte("event: patch\ndata: "))
            w.Write(msg)
            w.Write([]byte("\n\n"))
            flusher.Flush()
        case <-r.Context().Done():
            return
        }
    }
}

func main() {
    store := &FlagStore{
        flags:       make(map[string]*Flag),
        subscribers: make(map[chan []byte]struct{}),
    }

    http.HandleFunc("/api/flags", func(w http.ResponseWriter, r *http.Request) {
        switch r.Method {
        case http.MethodGet:
            json.NewEncoder(w).Encode(store.GetAll())
        case http.MethodPut:
            var flag Flag
            json.NewDecoder(r.Body).Decode(&flag)
            store.Update(&flag)
            w.WriteHeader(http.StatusOK)
        }
    })
    http.HandleFunc("/api/stream", store.StreamHandler)

    log.Println("Flag service starting on :8080")
    log.Fatal(http.ListenAndServe(":8080", nil))
}

# 渐进发布配置示例
flag: new-payment-gateway
type: release
rollout_plan:
  - stage: canary
    percentage: 1
    duration: 2h
    success_criteria:
      error_rate: "< 0.1%"
      p99_latency: "< 200ms"
    rollback_trigger:
      error_rate: "> 1%"

  - stage: early_adopters
    percentage: 10
    duration: 24h
    success_criteria:
      error_rate: "< 0.05%"
      conversion_rate: "> 95% of baseline"

  - stage: broad_rollout
    percentage: 50
    duration: 48h
    success_criteria:
      error_rate: "< 0.05%"
      all_metrics_nominal: true

  - stage: full_rollout
    percentage: 100
    monitoring_period: 7d

# 自动渐进发布控制器
import time
from dataclasses import dataclass
from typing import Optional

@dataclass
class RolloutStage:
    name: str
    percentage: int
    duration_hours: float
    max_error_rate: float
    max_p99_latency_ms: float

class ProgressiveRolloutController:
    def __init__(self, flag_client, metrics_client, flag_key: str):
        self.flag_client = flag_client
        self.metrics_client = metrics_client
        self.flag_key = flag_key

    def execute_rollout(self, stages: list[RolloutStage]) -> bool:
        for stage in stages:
            print(f"Starting stage: {stage.name} ({stage.percentage}%)")
            self.flag_client.set_percentage(self.flag_key, stage.percentage)

            # 等待指标稳定
            time.sleep(300)  # 5 分钟观察期

            # 持续监控
            check_interval = 60  # 每分钟检查一次
            total_checks = int(stage.duration_hours * 3600 / check_interval)

            for i in range(total_checks):
                metrics = self._get_current_metrics()

                if metrics["error_rate"] > stage.max_error_rate:
                    print(f"Error rate {metrics['error_rate']:.4f} exceeds "
                          f"threshold {stage.max_error_rate}. Rolling back.")
                    self._rollback()
                    return False

                if metrics["p99_latency_ms"] > stage.max_p99_latency_ms:
                    print(f"P99 latency {metrics['p99_latency_ms']:.1f}ms exceeds "
                          f"threshold {stage.max_p99_latency_ms}ms. Rolling back.")
                    self._rollback()
                    return False

                time.sleep(check_interval)

            print(f"Stage {stage.name} passed all checks.")

        print("Rollout complete. Flag is at 100%.")
        return True

    def _get_current_metrics(self) -> dict:
        return {
            "error_rate": self.metrics_client.query(
                f'rate(http_errors_total{{flag="{self.flag_key}"}}[5m])'
            ),
            "p99_latency_ms": self.metrics_client.query(
                f'histogram_quantile(0.99, '
                f'rate(http_duration_seconds_bucket{{flag="{self.flag_key}"}}[5m]))'
            ) * 1000
        }

    def _rollback(self):
        self.flag_client.set_percentage(self.flag_key, 0)
        print(f"Flag {self.flag_key} rolled back to 0%.")

{
  "key": "new-editor",
  "targeting_rules": [
    {
      "name": "内部员工优先",
      "priority": 1,
      "conditions": [
        { "attribute": "email", "op": "endsWith", "value": "@company.com" }
      ],
      "serve": { "variation": "enabled" }
    },
    {
      "name": "付费用户第二批",
      "priority": 2,
      "conditions": [
        { "attribute": "plan", "op": "in", "values": ["pro", "enterprise"] },
        { "attribute": "account_age_days", "op": ">=", "value": 90 }
      ],
      "serve": {
        "rollout": { "percentage": 30, "bucket_by": "org_id" }
      }
    },
    {
      "name": "免费用户最后",
      "priority": 3,
      "conditions": [
        { "attribute": "plan", "op": "eq", "value": "free" }
      ],
      "serve": {
        "rollout": { "percentage": 5, "bucket_by": "user_id" }
      }
    }
  ],
  "default": { "variation": "disabled" }
}

# 分层实验模型
class ExperimentLayer:
    """
    每个层将 100% 的流量分成不重叠的区间。
    同一层内的多个实验占据不同的区间，互斥。
    不同层之间正交（用不同的哈希盐）。
    """
    def __init__(self, layer_id: str, salt: str):
        self.layer_id = layer_id
        self.salt = salt
        self.experiments: list[Experiment] = []

    def assign_experiment(self, user_id: str) -> tuple[str, str] | None:
        bucket = self._hash_to_bucket(user_id)
        for exp in self.experiments:
            if exp.start_bucket <= bucket < exp.end_bucket:
                variant = exp.assign_variant(user_id, self.salt)
                return exp.name, variant
        return None

    def _hash_to_bucket(self, user_id: str) -> int:
        import hashlib
        h = hashlib.sha256(f"{self.salt}:{user_id}".encode()).hexdigest()
        return int(h[:8], 16) % 10000  # 0-9999，精度到 0.01%


class Experiment:
    def __init__(self, name: str, start_bucket: int, end_bucket: int,
                 variants: dict[str, int]):
        self.name = name
        self.start_bucket = start_bucket
        self.end_bucket = end_bucket
        self.variants = variants  # {"control": 5000, "treatment_a": 3000, "treatment_b": 2000}

    def assign_variant(self, user_id: str, layer_salt: str) -> str:
        import hashlib
        h = hashlib.sha256(
            f"{layer_salt}:{self.name}:{user_id}".encode()
        ).hexdigest()
        bucket = int(h[:8], 16) % 10000

        cumulative = 0
        for variant_name, weight in self.variants.items():
            cumulative += weight
            if bucket < cumulative:
                return variant_name
        return list(self.variants.keys())[-1]

experiment_layers:
  - id: "ui-layer"
    salt: "ui-2026q2"
    experiments:
      - name: "checkout-redesign"
        traffic: [0, 5000]      # 前 50% 的流量
        variants:
          control: 2500
          new_design: 2500
      - name: "search-ui-test"
        traffic: [5000, 8000]   # 50%-80% 的流量
        variants:
          control: 1500
          variant_a: 1500

  - id: "backend-layer"
    salt: "backend-2026q2"
    experiments:
      - name: "new-ranking-model"
        traffic: [0, 10000]     # 100% 的流量（和 UI 层正交）
        variants:
          control: 5000
          new_model: 5000

// 实验指标追踪（TypeScript）
interface ExperimentEvent {
    experimentKey: string;
    variantKey: string;
    userId: string;
    eventType: "exposure" | "conversion" | "custom";
    metricKey?: string;
    metricValue?: number;
    timestamp: number;
    properties?: Record<string, string | number>;
}

class ExperimentTracker {
    private exposureCache: Set<string> = new Set();

    trackExposure(experimentKey: string, variantKey: string, userId: string): void {
        // 去重：每个用户每个实验只记录一次曝光
        const dedupKey = `${experimentKey}:${userId}`;
        if (this.exposureCache.has(dedupKey)) {
            return;
        }
        this.exposureCache.add(dedupKey);

        this.emit({
            experimentKey,
            variantKey,
            userId,
            eventType: "exposure",
            timestamp: Date.now()
        });
    }

    trackConversion(
        experimentKey: string,
        variantKey: string,
        userId: string,
        metricKey: string,
        metricValue: number = 1
    ): void {
        this.emit({
            experimentKey,
            variantKey,
            userId,
            eventType: "conversion",
            metricKey,
            metricValue,
            timestamp: Date.now()
        });
    }

    private emit(event: ExperimentEvent): void {
        // 发送到事件收集服务（如 Kafka、Segment）
        eventBus.publish("experiment-events", event);
    }
}

// 使用示例
const tracker = new ExperimentTracker();
const variant = featureFlags.getVariant("pricing-experiment", user.id);

tracker.trackExposure("pricing-experiment", variant, user.id);

// 用户完成购买时
tracker.trackConversion(
    "pricing-experiment",
    variant,
    user.id,
    "purchase_completed",
    order.totalAmount
);

# 实验结果分析
import math

def z_test_proportions(
    control_conversions: int,
    control_total: int,
    treatment_conversions: int,
    treatment_total: int
) -> dict:
    """双比例 Z 检验"""
    p_control = control_conversions / control_total
    p_treatment = treatment_conversions / treatment_total
    p_pooled = (control_conversions + treatment_conversions) / (
        control_total + treatment_total
    )

    se = math.sqrt(
        p_pooled * (1 - p_pooled) * (1/control_total + 1/treatment_total)
    )

    if se == 0:
        return {"z_score": 0, "p_value": 1.0, "significant": False}

    z_score = (p_treatment - p_control) / se

    # 双尾检验 p 值（近似计算）
    from statistics import NormalDist
    p_value = 2 * (1 - NormalDist().cdf(abs(z_score)))

    return {
        "control_rate": p_control,
        "treatment_rate": p_treatment,
        "lift": (p_treatment - p_control) / p_control if p_control > 0 else 0,
        "z_score": z_score,
        "p_value": p_value,
        "significant": p_value < 0.05,  # 95% 置信水平
        "sample_size": control_total + treatment_total
    }

# 示例：购买按钮颜色 A/B 测试结果
result = z_test_proportions(
    control_conversions=1200,    # 蓝色按钮：1200 次转化
    control_total=50000,         # 蓝色按钮：50000 次曝光
    treatment_conversions=1350,  # 绿色按钮：1350 次转化
    treatment_total=50000        # 绿色按钮：50000 次曝光
)

# 输出：
# control_rate: 0.024 (2.4%)
# treatment_rate: 0.027 (2.7%)
# lift: 0.125 (12.5%)
# p_value: 0.003
# significant: True

// JUnit 5 参数化测试示例
@ParameterizedTest
@CsvSource({
    "true, 期望新流程结果",
    "false, 期望旧流程结果"
})
void testCheckoutFlow(boolean newCheckoutEnabled, String expectedBehavior) {
    // 设置 flag 状态
    flagClient.override("new-checkout-flow", newCheckoutEnabled);

    CheckoutResult result = checkoutService.process(testOrder);

    if (newCheckoutEnabled) {
        assertThat(result.getFlowType()).isEqualTo("new");
    } else {
        assertThat(result.getFlowType()).isEqualTo("legacy");
    }
}

# 使用 allpairspy 生成成对测试用例
from allpairspy import AllPairs

flags = {
    "new_checkout": [True, False],
    "new_payment":  [True, False],
    "new_search":   [True, False],
    "dark_mode":    [True, False],
    "new_pricing":  [True, False],
}

parameters = list(flags.values())
flag_names = list(flags.keys())

print(f"完全组合: {2**len(flags)} = {2**len(flags)} 个测试用例")

test_cases = list(AllPairs(parameters))
print(f"成对测试: {len(test_cases)} 个测试用例")

for i, case in enumerate(test_cases):
    config = dict(zip(flag_names, case))
    print(f"测试用例 {i+1}: {config}")

# 输出示例：
# 完全组合: 32 个测试用例
# 成对测试: 6 个测试用例
# 测试用例 1: {new_checkout: True, new_payment: True, new_search: True, ...}
# 测试用例 2: {new_checkout: True, new_payment: False, new_search: False, ...}
# ...

# 测试环境 flag 配置
environments:
  unit_test:
    description: "单元测试：所有新 flag 开启，测试新代码路径"
    flags:
      new-checkout-flow: true
      new-payment-gateway: true
      new-search-algorithm: true

  integration_test:
    description: "集成测试：混合状态，模拟真实灰度场景"
    flags:
      new-checkout-flow: true
      new-payment-gateway: false  # 旧支付通道
      new-search-algorithm: true

  regression_test:
    description: "回归测试：所有新 flag 关闭，确保旧路径不被破坏"
    flags:
      new-checkout-flow: false
      new-payment-gateway: false
      new-search-algorithm: false

  production_mirror:
    description: "生产镜像：使用与生产完全相同的 flag 状态"
    source: production  # 从生产环境同步 flag 状态

// Go 测试中的 flag 覆盖
type FlagOverride struct {
    t       *testing.T
    client  *FlagClient
    overrides map[string]interface{}
    originals map[string]interface{}
}

func WithFlags(t *testing.T, client *FlagClient) *FlagOverride {
    return &FlagOverride{
        t:         t,
        client:    client,
        overrides: make(map[string]interface{}),
        originals: make(map[string]interface{}),
    }
}

func (fo *FlagOverride) Set(key string, value interface{}) *FlagOverride {
    fo.originals[key] = fo.client.GetRawValue(key)
    fo.client.SetOverride(key, value)
    fo.overrides[key] = value
    return fo
}

func (fo *FlagOverride) Cleanup() {
    for key, original := range fo.originals {
        fo.client.SetOverride(key, original)
    }
}

// 使用方式
func TestNewCheckout(t *testing.T) {
    flags := WithFlags(t, flagClient).
        Set("new-checkout-flow", true).
        Set("new-payment-gateway", true)
    defer flags.Cleanup()

    result := checkoutService.Process(testOrder)
    assert.Equal(t, "new", result.FlowType)
}

# Flag 元数据规范
flag:
  key: "new-checkout-flow"
  name: "新结账流程"
  description: "使用重构后的三步结账流程替代当前的五步流程"

  # 分类与生命周期
  type: release          # release | experiment | ops | permission
  status: active         # draft | active | complete | deprecated
  created_at: "2026-03-01"
  expected_removal: "2026-04-15"

  # 所有权
  owner_team: "checkout-team"
  owner_person: "alice@company.com"
  jira_ticket: "CHECKOUT-1234"

  # 技术信息
  code_references:
    - file: "src/checkout/service.go"
      line: 42
    - file: "src/checkout/handler.go"
      line: 78
  sdk_languages: ["go", "typescript"]

  # 风险评估
  risk_level: medium     # low | medium | high | critical
  rollback_plan: "将 flag 设为 false 即可回退到旧流程"
  dependencies:
    - "payment-service-v2"

  # 审计
  last_modified_by: "bob@company.com"
  last_modified_at: "2026-03-20"
  change_history:
    - date: "2026-03-01"
      action: "created"
      by: "alice@company.com"
    - date: "2026-03-10"
      action: "enabled for internal"
      by: "alice@company.com"
    - date: "2026-03-20"
      action: "rollout to 10%"
      by: "bob@company.com"

{
  "audit_event": {
    "id": "evt-2026-0401-001",
    "timestamp": "2026-04-01T10:30:00Z",
    "actor": {
      "user": "alice@company.com",
      "ip": "10.0.1.42",
      "role": "release-manager"
    },
    "action": "flag.targeting.updated",
    "resource": {
      "flag_key": "new-payment-gateway",
      "environment": "production"
    },
    "changes": {
      "before": { "percentage": 10 },
      "after": { "percentage": 50 }
    },
    "approval": {
      "required": true,
      "approved_by": "bob@company.com",
      "approval_ticket": "CHANGE-5678"
    },
    "context": {
      "reason": "Phase 2 rollout after 48h observation",
      "related_incidents": []
    }
  }
}

# GitHub Flipper 使用模式（简化示例）
class CheckoutController < ApplicationController
  def create
    if feature_enabled?(:new_checkout_flow, current_user)
      new_checkout_flow
    else
      legacy_checkout_flow
    end
  end

  private

  def new_checkout_flow
    # ...
  end

  def legacy_checkout_flow
    # ...
  end
end

# 扫描超期未清理的 flag
import datetime

def find_stale_flags(flag_store, grace_period_days: int = 14) -> list[dict]:
    stale_flags = []
    now = datetime.datetime.now()

    for flag in flag_store.get_all_flags():
        if flag.type not in ("release", "experiment"):
            continue  # 运维和权限 flag 不适用自动过期

        if flag.expected_removal is None:
            stale_flags.append({
                "key": flag.key,
                "reason": "没有设置预期移除日期",
                "owner": flag.owner,
                "created": flag.created_at,
                "age_days": (now - flag.created_at).days
            })
            continue

        deadline = flag.expected_removal + datetime.timedelta(days=grace_period_days)
        if now > deadline:
            overdue_days = (now - flag.expected_removal).days
            stale_flags.append({
                "key": flag.key,
                "reason": f"超过预期移除日期 {overdue_days} 天",
                "owner": flag.owner,
                "expected_removal": flag.expected_removal,
                "age_days": (now - flag.created_at).days
            })

    return sorted(stale_flags, key=lambda x: x["age_days"], reverse=True)

-- 查找过去 30 天内求值结果从未变化的 flag（始终返回同一个值）
SELECT
    flag_key,
    MIN(variation) AS min_variation,
    MAX(variation) AS max_variation,
    COUNT(DISTINCT variation) AS distinct_variations,
    COUNT(*) AS total_evaluations,
    MIN(timestamp) AS first_seen,
    MAX(timestamp) AS last_seen
FROM flag_evaluations
WHERE timestamp > NOW() - INTERVAL '30 days'
GROUP BY flag_key
HAVING COUNT(DISTINCT variation) = 1
ORDER BY total_evaluations DESC;

#!/bin/bash
# 扫描代码中引用的 flag，与 flag 服务中的定义做交叉对比

# 从代码中提取所有 flag 引用
grep -rn 'isEnabled\|getVariant\|feature_enabled' src/ \
    | grep -oP '"[a-z][a-z0-9-]+"' \
    | sort -u > flags_in_code.txt

# 从 flag 服务获取所有已定义的 flag
curl -s http://flag-service/api/flags \
    | jq -r '.[].key' \
    | sort -u > flags_in_service.txt

# 找出代码中引用但服务中不存在的 flag（幽灵引用）
comm -23 flags_in_code.txt flags_in_service.txt > orphan_references.txt

# 找出服务中存在但代码中未引用的 flag（僵尸 flag）
comm -13 flags_in_code.txt flags_in_service.txt > zombie_flags.txt

echo "幽灵引用（代码引用了不存在的 flag）："
cat orphan_references.txt

echo "僵尸 Flag（服务中存在但代码未引用）："
cat zombie_flags.txt

# 自动 flag 清理工具（简化版）
import ast
import sys

class FlagCleanupTransformer(ast.NodeTransformer):
    """
    将 flag 判断代码替换为确定的分支。
    例如：if flag.is_enabled("x"):
              do_new()
          else:
              do_old()
    当 flag "x" 的最终值为 True 时，替换为：
          do_new()
    """

    def __init__(self, flag_key: str, final_value: bool):
        self.flag_key = flag_key
        self.final_value = final_value
        self.changes = []

    def visit_If(self, node: ast.If) -> ast.AST:
        self.generic_visit(node)

        if self._is_flag_check(node.test):
            self.changes.append({
                "line": node.lineno,
                "action": "removed flag check",
                "flag": self.flag_key
            })

            if self.final_value:
                # 保留 if 分支的代码
                return node.body
            else:
                # 保留 else 分支的代码
                return node.orelse if node.orelse else []

        return node

    def _is_flag_check(self, node: ast.expr) -> bool:
        if isinstance(node, ast.Call):
            if isinstance(node.func, ast.Attribute):
                if node.func.attr in ("is_enabled", "isEnabled"):
                    if node.args and isinstance(node.args[0], ast.Constant):
                        return node.args[0].value == self.flag_key
        return False

{
  "piranha": [
    {
      "methodName": "isEnabled",
      "flagType": "treated",
      "argumentIndex": 0
    },
    {
      "methodName": "getVariant",
      "flagType": "treated",
      "argumentIndex": 0
    }
  ],
  "flags_to_clean": [
    {
      "flag_name": "new-checkout-flow",
      "treated_value": true
    },
    {
      "flag_name": "old-search-algorithm",
      "treated_value": false
    }
  ]
}

# CI/CD 管道示例（GitHub Actions）
name: Deploy and Release
on:
  push:
    branches: [main]

jobs:
  deploy:
    runs-on: ubuntu-latest
    steps:
      - name: Build
        run: make build

      - name: Run tests (all flags on)
        run: make test FLAG_ENV=all_on

      - name: Run tests (all flags off)
        run: make test FLAG_ENV=all_off

      - name: Deploy to production
        run: make deploy
        # 新功能已部署，但 flag 关闭，用户看不到

  release:
    needs: deploy
    runs-on: ubuntu-latest
    steps:
      - name: Enable flag for internal users
        run: |
          curl -X PATCH \
            -H "Authorization: Bearer ${{ secrets.FLAG_API_TOKEN }}" \
            -H "Content-Type: application/json" \
            -d '{"targets": [{"variation": 0, "values": ["internal-segment"]}]}' \
            "https://flag-service.internal/api/flags/new-feature/targeting"

      - name: Smoke test with internal users
        run: make smoke-test USER_SEGMENT=internal

      - name: Gradual rollout to 10%
        run: |
          curl -X PATCH \
            -H "Authorization: Bearer ${{ secrets.FLAG_API_TOKEN }}" \
            -H "Content-Type: application/json" \
            -d '{"fallthrough": {"rollout": {"variations": [{"variation": 0, "weight": 10000}, {"variation": 1, "weight": 90000}]}}}' \
            "https://flag-service.internal/api/flags/new-feature/targeting"

# 部署前置条件检查
def pre_deploy_flag_check(flag_client, environment: str) -> bool:
    checks = [
        # 确保降级开关处于正常状态
        ("ops.recommendation.circuit-breaker", True, "推荐服务降级开关应该开启"),
        # 确保上一个版本的发布 flag 已经全量
        ("release.checkout.v2", True, "Checkout V2 应该已经全量上线"),
    ]

    all_passed = True
    for flag_key, expected_value, message in checks:
        actual = flag_client.is_enabled(flag_key)
        if actual != expected_value:
            print(f"FAIL: {message} (expected={expected_value}, actual={actual})")
            all_passed = False
        else:
            print(f"PASS: {flag_key} = {actual}")

    return all_passed

// OpenFeature 统一 API（Java）
import dev.openfeature.sdk.OpenFeatureAPI;
import dev.openfeature.sdk.Client;
import dev.openfeature.sdk.EvaluationContext;

// 注册 Provider（可以是 LaunchDarkly、Unleash、或自建系统）
OpenFeatureAPI api = OpenFeatureAPI.getInstance();
api.setProvider(new LaunchDarklyProvider("sdk-key"));
// 或者：api.setProvider(new UnleashProvider(unleashConfig));
// 或者：api.setProvider(new CustomProvider(myFlagService));

// 获取客户端
Client client = api.getClient();

// 使用统一 API 求值——不依赖任何特定供应商
EvaluationContext ctx = new ImmutableContext("user-123",
    Map.of("plan", new Value("enterprise"),
           "country", new Value("CN")));

boolean showNewFeature = client.getBooleanValue(
    "new-checkout-flow",
    false,  // 默认值
    ctx
);

String buttonColor = client.getStringValue(
    "checkout-button-color",
    "blue",
    ctx
);

// OpenFeature Hook 示例（TypeScript）
import { Hook, HookContext, EvaluationDetails } from "@openfeature/server-sdk";

const loggingHook: Hook = {
    before(hookContext: HookContext): void {
        console.log(`Evaluating flag: ${hookContext.flagKey}`);
    },

    after(
        hookContext: HookContext,
        evaluationDetails: EvaluationDetails<any>
    ): void {
        console.log(
            `Flag ${hookContext.flagKey} evaluated to ${evaluationDetails.value} ` +
            `(reason: ${evaluationDetails.reason})`
        );
    },

    error(hookContext: HookContext, error: Error): void {
        console.error(
            `Flag ${hookContext.flagKey} evaluation failed: ${error.message}`
        );
    }
};

// 注册全局 Hook
OpenFeature.addHooks(loggingHook);

// 反模式：flag 嵌套
if (flags.isEnabled("feature-a")) {
    if (flags.isEnabled("feature-b")) {
        if (flags.isEnabled("feature-c")) {
            // 这三个 flag 的组合有 8 种状态
            // 你确定都测试过了？
            doSomethingComplicated();
        }
    }
}

# 反模式：用 flag 传递配置值
timeout = flag_client.get_integer("service-timeout-ms", 5000)
max_retries = flag_client.get_integer("max-retries", 3)
batch_size = flag_client.get_integer("batch-size", 100)

// 这个 flag 三年前就应该删除了
if flags.IsEnabled("new-user-onboarding-v2") {
    // "v2" 已经是唯一的版本了
    // 没有人知道 "v1" 长什么样
    // 没有人敢删除这个判断
    showOnboarding()
}

// 反模式：每次求值都调用远程服务
async function handleRequest(req) {
    // 每个请求都发一次 HTTP 调用，延迟 10-50ms
    const enabled = await fetch("https://flag-service/api/evaluate", {
        method: "POST",
        body: JSON.stringify({ key: "my-flag", context: req.user })
    }).then(r => r.json());

    if (enabled) {
        return newHandler(req);
    }
    return legacyHandler(req);
}