feat(feature-flag): Redis Feature Flag 灰度发布基础设施

Redis Feature Flag 灰度发布基础设施：白名单+百分比切流+全局开关，热更新，内部管理API

apps/worker/video_processing/render_engine_resolver.py

@@ -0,0 +1,204 @@
+"""渲染引擎 Feature Flag 解析器。
+
+封装渲染引擎选择逻辑，支持：
+- 环境变量作为默认值（RENDER_ENGINE=legacy/unified）
+- Redis Feature Flag 运行时覆盖（白名单 + 百分比 + 全局开关）
+- 定时刷新，支持热更新不重启 worker
+
+使用方式：
+    resolver = RenderEngineResolver(redis_url="redis://...", default_engine="legacy")
+    engine = resolver.get_engine(user_id="user123")
+    # engine: "legacy" 或 "unified"
+"""
+
+from __future__ import annotations
+
+import logging
+import threading
+from typing import Optional
+
+from packages.adapters.redis.feature_flag_store import (
+    FeatureFlagConfig,
+    FeatureFlagStore,
+    InMemoryFeatureFlagStore,
+    RedisFeatureFlagStore,
+)
+
+logger = logging.getLogger(__name__)
+
+# Feature Flag 名称常量
+FLAG_RENDER_ENGINE = "render_engine"
+
+# 引擎常量
+ENGINE_LEGACY = "legacy"
+ENGINE_UNIFIED = "unified"
+VALID_ENGINES = {ENGINE_LEGACY, ENGINE_UNIFIED}
+
+
+class RenderEngineResolver:
+    """渲染引擎选择器。
+
+    判定逻辑（从高到低）：
+    1. Redis flag 白名单匹配 → unified
+    2. Redis flag 百分比命中 → unified
+    3. Redis flag 全局开启（100%）→ unified
+    4. 环境变量默认值 → legacy / unified
+
+    当 Redis 不可用时，自动降级到环境变量默认值，不影响业务。
+    """
+
+    def __init__(
+        self,
+        default_engine: str = ENGINE_LEGACY,
+        redis_url: Optional[str] = None,
+        refresh_interval: float = 30.0,
+        store: Optional[FeatureFlagStore] = None,
+    ) -> None:
+        """
+        Args:
+            default_engine: 环境变量默认的引擎名（legacy / unified）
+            redis_url: Redis 连接 URL，传 None 时使用内存实现（测试用）
+            refresh_interval: Redis flag 配置刷新间隔（秒）
+            store: 直接传入 store 实例（测试用，优先级高于 redis_url）
+        """
+        self._default_engine = default_engine.lower() if default_engine else ENGINE_LEGACY
+        if self._default_engine not in VALID_ENGINES:
+            logger.warning(
+                "Invalid default engine '%s', fallback to '%s'",
+                self._default_engine,
+                ENGINE_LEGACY,
+            )
+            self._default_engine = ENGINE_LEGACY
+
+        if store is not None:
+            self._store = store
+        elif redis_url:
+            self._store = RedisFeatureFlagStore(redis_url=redis_url)
+        else:
+            self._store = InMemoryFeatureFlagStore()
+            logger.info("No Redis configured, using in-memory feature flag store")
+
+        self._refresh_interval = refresh_interval
+        self._lock = threading.Lock()
+        self._cached_config: Optional[FeatureFlagConfig] = None
+        self._last_refresh: float = 0.0
+
+    def _maybe_refresh(self) -> None:
+        """惰性刷新配置，超过刷新间隔时从存储重新读取。"""
+        import time
+
+        now = time.time()
+        if now - self._last_refresh < self._refresh_interval:
+            return
+
+        try:
+            config = self._store.get(FLAG_RENDER_ENGINE)
+            with self._lock:
+                self._cached_config = config
+                self._last_refresh = now
+        except Exception as exc:
+            logger.warning("Failed to refresh render engine flag: %s", exc)
+            # 刷新失败时保留旧缓存，不中断业务
+            if self._cached_config is None:
+                # 首次就读失败，设一个默认值
+                with self._lock:
+                    self._cached_config = FeatureFlagConfig(name=FLAG_RENDER_ENGINE)
+                    self._last_refresh = now
+
+    def _get_config(self) -> FeatureFlagConfig:
+        """获取当前 flag 配置（带缓存）。"""
+        if self._cached_config is None:
+            self._maybe_refresh()
+        else:
+            self._maybe_refresh()
+        return self._cached_config or FeatureFlagConfig(name=FLAG_RENDER_ENGINE)
+
+    def get_engine(self, user_id: Optional[str] = None) -> str:
+        """获取当前应该使用的渲染引擎。
+
+        Args:
+            user_id: 用户ID，用于白名单匹配和百分比哈希。
+                    传 None 时只看全局开关。
+
+        Returns:
+            "legacy" 或 "unified"
+        """
+        config = self._get_config()
+
+        # 全局关闭 → 用默认值
+        if not config.enabled:
+            return self._default_engine
+
+        # 白名单匹配 / 百分比命中 → unified
+        if config.is_active(user_id):
+            return ENGINE_UNIFIED
+
+        # 未命中灰度 → 用默认值
+        return self._default_engine
+
+    def should_use_unified(self, user_id: Optional[str] = None) -> bool:
+        """便捷方法：是否应该使用统一渲染引擎。"""
+        return self.get_engine(user_id) == ENGINE_UNIFIED
+
+    def force_refresh(self) -> None:
+        """强制立即刷新配置（用于管理接口修改后立即生效）。"""
+        self._last_refresh = 0.0
+        if isinstance(self._store, RedisFeatureFlagStore):
+            self._store.invalidate_cache(FLAG_RENDER_ENGINE)
+        self._maybe_refresh()
+
+    def get_config_snapshot(self) -> dict:
+        """获取当前配置快照（用于管理接口展示）。"""
+        config = self._get_config()
+        return {
+            "flag_name": FLAG_RENDER_ENGINE,
+            "default_engine": self._default_engine,
+            "enabled": config.enabled,
+            "percentage": config.percentage,
+            "whitelist": sorted(config.whitelist),
+            "refresh_interval": self._refresh_interval,
+            "last_refresh": self._last_refresh,
+        }
+
+    def set_flag(self, config: FeatureFlagConfig) -> None:
+        """设置 flag 配置（管理接口用）。"""
+        config.name = FLAG_RENDER_ENGINE
+        self._store.set(config)
+        self.force_refresh()
+
+
+# 全局单例
+_resolver: Optional[RenderEngineResolver] = None
+_resolver_lock = threading.Lock()
+
+
+def get_render_engine_resolver() -> RenderEngineResolver:
+    """获取全局单例（基于 worker 配置）。"""
+    global _resolver
+    if _resolver is not None:
+        return _resolver
+
+    with _resolver_lock:
+        if _resolver is not None:
+            return _resolver
+
+        try:
+            from worker_app.core.config import get_settings
+
+            settings = get_settings()
+            redis_url = getattr(settings, "redis_url", None) or getattr(settings, "broker_url", None)
+            default = getattr(settings, "render_engine", ENGINE_LEGACY)
+            _resolver = RenderEngineResolver(
+                default_engine=default,
+                redis_url=redis_url,
+            )
+            logger.info(
+                "RenderEngineResolver initialized: default=%s, redis=%s",
+                default,
+                bool(redis_url),
+            )
+        except Exception as exc:
+            logger.warning("Failed to init RenderEngineResolver from settings: %s", exc)
+            _resolver = RenderEngineResolver(default_engine=ENGINE_LEGACY)
+
+    return _resolver