From 7e6ccc726a84b37f47a312e5e5eeb7f8d1511485 Mon Sep 17 00:00:00 2001 From: zero Date: Wed, 24 Jun 2026 08:14:10 -0400 Subject: [PATCH 1/5] Remove stray EXIT: deploy-log file (invalid path on Windows) --- EXIT: | 37 ------------------------------------- 1 file changed, 37 deletions(-) delete mode 100644 EXIT: diff --git a/EXIT: b/EXIT: deleted file mode 100644 index 8d246def..00000000 --- a/EXIT: +++ /dev/null @@ -1,37 +0,0 @@ - -=================================================== -WaddleBot Beta Deployment Script -=================================================== - -[INFO] Tag: beta-1773154482 -[INFO] Method: helm -[INFO] Namespace: waddlebot -[INFO] Kube Context: dal2-beta - -=================================================== -Checking Prerequisites -=================================================== - -[SUCCESS] Docker found: Docker version 28.2.2, build 28.2.2-0ubuntu1 -[SUCCESS] kubectl found: installed -[SUCCESS] Helm found: v4.1.1+g5caf004 -[SUCCESS] WaddleBot project directory verified -[SUCCESS] Kubernetes context: dal2-beta -[INFO] Loaded NPM_TOKEN from ~/code/.gh-token -[SUCCESS] NPM_TOKEN configured - -=================================================== -Skipping Image Build (--skip-build flag set) -=================================================== - - -=================================================== -Deploying to Beta Cluster with Helm -=================================================== - -[INFO] Checking if namespace waddlebot exists... -[SUCCESS] Namespace waddlebot already exists -[INFO] Deploying WaddleBot to beta cluster... -level=WARN msg="upgrade failed" name=waddlebot error="resource Deployment/waddlebot/waddlebot-analytics-core not ready. status: Failed, message: Progress deadline exceeded" -Error: UPGRADE FAILED: resource Deployment/waddlebot/waddlebot-analytics-core not ready. status: Failed, message: Progress deadline exceeded -[ERROR] Helm deployment failed From 954201173eb982e3aed87c22866fc77767d75436 Mon Sep 17 00:00:00 2001 From: zero Date: Wed, 8 Jul 2026 10:33:09 -0400 Subject: [PATCH 2/5] Add feature_flags and feature_flag_audit tables (migration 068) Scoped flags: community_id NULL = global default, platform NULL = all platforms, rollout_pct 0-100 for sticky percentage rollouts. Append-only audit table. Verified against a live PostgreSQL 16 in a rolled-back transaction; picked up automatically by the Alembic baseline glob. Co-Authored-By: Claude Fable 5 --- .../postgres/migrations/068_feature_flags.sql | 52 +++++++++++++++++++ docs/architecture/table-ownership.md | 2 + 2 files changed, 54 insertions(+) create mode 100644 config/postgres/migrations/068_feature_flags.sql diff --git a/config/postgres/migrations/068_feature_flags.sql b/config/postgres/migrations/068_feature_flags.sql new file mode 100644 index 00000000..182caf5d --- /dev/null +++ b/config/postgres/migrations/068_feature_flags.sql @@ -0,0 +1,52 @@ +-- Migration 068: Feature flag system +-- Adds scoped feature flags (global / per-community / per-platform) with +-- percentage-based rollout, plus an append-only audit trail of flag changes. + +CREATE TABLE IF NOT EXISTS feature_flags ( + id SERIAL PRIMARY KEY, + flag_key VARCHAR(100) NOT NULL, -- dot-namespaced, e.g. 'module.loyalty_interaction' + community_id INTEGER REFERENCES communities(id) ON DELETE CASCADE, -- NULL = global default + platform VARCHAR(50), -- NULL = all platforms (e.g. 'twitch','discord','slack') + is_enabled BOOLEAN NOT NULL DEFAULT false, + rollout_pct SMALLINT NOT NULL DEFAULT 100 CHECK (rollout_pct BETWEEN 0 AND 100), + description TEXT, + updated_by VARCHAR(255), + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP +); + +COMMENT ON TABLE feature_flags IS 'Scoped feature flags with percentage rollout. Scope resolves most-specific-wins: (community + platform) > community > platform > global.'; +COMMENT ON COLUMN feature_flags.flag_key IS 'Dot-namespaced flag identifier, e.g. ''module.loyalty_interaction''.'; +COMMENT ON COLUMN feature_flags.community_id IS 'Owning community; NULL = global default (same convention as commands table).'; +COMMENT ON COLUMN feature_flags.platform IS 'Platform scope; NULL = all platforms.'; +COMMENT ON COLUMN feature_flags.rollout_pct IS 'Percentage of the scoped audience the flag is active for (0-100).'; + +-- Uniqueness across the nullable scope columns (flag_key + community + platform). +-- COALESCE sentinels let NULL community/platform participate in the unique constraint. +CREATE UNIQUE INDEX IF NOT EXISTS idx_feature_flags_scope + ON feature_flags (flag_key, COALESCE(community_id, -1), COALESCE(platform, '*')); + +-- Lookup index for flag resolution by key + community scope. +CREATE INDEX IF NOT EXISTS idx_feature_flags_lookup + ON feature_flags (flag_key, community_id); + +CREATE TABLE IF NOT EXISTS feature_flag_audit ( + id SERIAL PRIMARY KEY, + flag_key VARCHAR(100) NOT NULL, + community_id INTEGER, -- no FK: audit history is retained even if the community is deleted + platform VARCHAR(50), + action VARCHAR(20) NOT NULL, -- 'created', 'updated', 'deleted' + old_value JSONB, + new_value JSONB, + changed_by VARCHAR(255), + changed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP +); + +COMMENT ON TABLE feature_flag_audit IS 'Append-only audit trail of feature flag changes. Retained independently of feature_flags rows.'; + +-- Audit history lookup by flag, newest first. +CREATE INDEX IF NOT EXISTS idx_feature_flag_audit_flag_key + ON feature_flag_audit (flag_key, changed_at); + +-- Analyze +ANALYZE feature_flags; diff --git a/docs/architecture/table-ownership.md b/docs/architecture/table-ownership.md index 309cb9a1..b4425d3d 100644 --- a/docs/architecture/table-ownership.md +++ b/docs/architecture/table-ownership.md @@ -216,6 +216,8 @@ Complete inventory of all tables in WaddleBot, organized by owning module, with | `user_presence_settings` | Presence Sync | `mod_core_engagement` | All | Engagement | — | 060 | | `presence_events_log` | Presence Sync | `mod_core_engagement` | All | Engagement | — | 060 | | `data_deletion_requests` | GDPR (Privacy) | `hub_admin` | Hub Admin, Security Core | Privacy | — | 062 | +| `feature_flags` | Feature Flags | `hub_admin` | All | Hub Admin | — | 068 | +| `feature_flag_audit` | Feature Flags | `hub_admin` | Hub Admin, Security Core | Hub Admin | — | 068 | --- From 646e5e6709bc5a64639012f0820356850d5e66b1 Mon Sep 17 00:00:00 2001 From: zero Date: Wed, 8 Jul 2026 10:33:09 -0400 Subject: [PATCH 3/5] Add FeatureFlagService evaluation helper to flask_core Async Redis-cached flag resolution (most-specific scope wins), sticky sha256 percentage bucketing on community_id, fail-open on errors, and pub/sub invalidation via the feature_flags:reload channel. Mirrored into the services/core-community flask_core copy per the dual-layout convention. 20 stdlib-only unit tests. Co-Authored-By: Claude Fable 5 --- libs/flask_core/flask_core/__init__.py | 4 + libs/flask_core/flask_core/feature_flags.py | 313 ++++++++++++++++++ .../libs/flask_core/flask_core/__init__.py | 4 + .../flask_core/flask_core/feature_flags.py | 313 ++++++++++++++++++ tests/unit/test_feature_flags.py | 284 ++++++++++++++++ 5 files changed, 918 insertions(+) create mode 100644 libs/flask_core/flask_core/feature_flags.py create mode 100644 services/core-community/libs/flask_core/flask_core/feature_flags.py create mode 100644 tests/unit/test_feature_flags.py diff --git a/libs/flask_core/flask_core/__init__.py b/libs/flask_core/flask_core/__init__.py index 2ad62967..5753cbfe 100644 --- a/libs/flask_core/flask_core/__init__.py +++ b/libs/flask_core/flask_core/__init__.py @@ -36,6 +36,7 @@ record_request_metrics ) from .cache import CacheManager, create_cache_manager +from .feature_flags import FeatureFlagService, create_feature_flag_service from .rate_limiter import RateLimiter, RateLimitExceeded, create_rate_limiter from .message_queue import MessageQueue, Message, create_message_queue from .stream_pipeline import StreamPipeline, StreamEvent, create_stream_pipeline @@ -148,6 +149,9 @@ # Cache "CacheManager", "create_cache_manager", + # Feature Flags + "FeatureFlagService", + "create_feature_flag_service", # Rate Limiting "RateLimiter", "RateLimitExceeded", diff --git a/libs/flask_core/flask_core/feature_flags.py b/libs/flask_core/flask_core/feature_flags.py new file mode 100644 index 00000000..a2d434b9 --- /dev/null +++ b/libs/flask_core/flask_core/feature_flags.py @@ -0,0 +1,313 @@ +""" +Feature-Flag Evaluation Service +================================ + +Shared, Redis-cached feature-flag evaluator for all Waddles modules. + +Backing table (migration 068):: + + feature_flags(id, flag_key, community_id NULL, platform NULL, + is_enabled, rollout_pct, description, updated_by, + created_at, updated_at) + UNIQUE (flag_key, COALESCE(community_id, -1), COALESCE(platform, '*')) + +Resolution picks the *most specific* matching row for a +``(flag_key, community_id, platform)`` lookup, in this order (first match +wins):: + + 1. (community_id, platform) -- exact community + exact platform + 2. (community_id, NULL) -- exact community, all platforms + 3. (NULL, platform) -- global community, exact platform + 4. (NULL, NULL) -- global default + +Fail-open semantics: when no row matches, or when Redis / the DB raise, the +caller-supplied ``default`` is returned (an absent flag means the feature is +on). A matched row with ``is_enabled = false`` is an explicit kill-switch and +returns ``False``. When ``is_enabled = true`` and ``rollout_pct < 100`` a +deterministic, sticky bucketing decides the result so the same +``(flag_key, community_id)`` always lands the same way. + +Redis-client vs CacheManager +---------------------------- +This service takes a **raw** ``redis.asyncio`` client (the same object the +Router already holds and the object ``AiChatterConfigCache`` consumes), not a +``flask_core.CacheManager``. Rationale: the Router and the platform receivers +pass their bare Redis connection around directly, and this service needs +``PUBLISH`` + ``SCAN`` for reload fan-out / cache invalidation -- surface that +``CacheManager`` does not expose. A raw client is therefore the lowest common +denominator that *both* the Router and other ``flask_core`` consumers already +have on hand, so no adapter is required at any call site. The ``__init__`` +signature stays ``(redis_client, dal, default_ttl)`` regardless. +""" + +from __future__ import annotations + +import hashlib +import json +import logging +from dataclasses import dataclass +from typing import Any + +logger = logging.getLogger(__name__) + +#: Redis Pub/Sub channel used to broadcast cache-invalidation reloads. +RELOAD_CHANNEL = "feature_flags:reload" + + +@dataclass(slots=True) +class _FlagRow: + """A single ``feature_flags`` row materialized for specificity ranking.""" + + community_id: int | None + platform: str | None + is_enabled: bool + rollout_pct: int + + +class FeatureFlagService: + """ + Evaluate feature flags with a most-specific-wins resolution and a + Redis-cached final boolean. + + All public evaluation paths are fail-open: any Redis or DB error is logged + at ``warning`` level and the caller-supplied ``default`` is returned. + """ + + def __init__(self, redis_client: Any, dal: Any, default_ttl: int = 300) -> None: + """ + Args: + redis_client: Raw ``redis.asyncio`` client (may be ``None`` -- the + service then falls straight through to the DB and never caches). + dal: Object exposing a synchronous ``executesql(sql, params)`` that + returns a list of row tuples (PyDAL ``AsyncDAL`` / DAL). + default_ttl: TTL in seconds for cached boolean results. + """ + self.redis = redis_client + self.dal = dal + self.default_ttl = default_ttl + + # ------------------------------------------------------------------ # + # Key helpers + # ------------------------------------------------------------------ # + @staticmethod + def _cache_key( + flag_key: str, community_id: int | None, platform: str | None + ) -> str: + scope = community_id if community_id is not None else "global" + return f"feature_flag:{scope}:{flag_key}:{platform or '*'}" + + @staticmethod + def _invalidate_pattern(flag_key: str) -> str: + # Matches every community/platform variant cached for this flag. + return f"feature_flag:*:{flag_key}:*" + + @staticmethod + def _bucket_enabled( + flag_key: str, community_id: int | None, rollout_pct: int + ) -> bool: + """Deterministic sticky bucketing: same inputs -> same decision.""" + scope = community_id if community_id is not None else "global" + digest = hashlib.sha256(f"{flag_key}:{scope}".encode()).hexdigest() + return int(digest, 16) % 100 < rollout_pct + + # ------------------------------------------------------------------ # + # Public API + # ------------------------------------------------------------------ # + async def is_enabled( + self, + flag_key: str, + community_id: int | None = None, + platform: str | None = None, + *, + default: bool = True, + ) -> bool: + """ + Resolve whether ``flag_key`` is enabled for the given scope. + + Checks the Redis cache first; on a miss, resolves from the DB (one + parameterized query fetching every row for ``flag_key`` scoped to the + community, then picks specificity in Python) and caches the boolean. + + Returns ``default`` when no row matches or on any Redis/DB failure. + """ + cache_key = self._cache_key(flag_key, community_id, platform) + + # 1) Cache fast-path. + try: + if self.redis is not None: + cached = await self.redis.get(cache_key) + if cached is not None: + return self._decode_bool(cached) + except Exception as e: # noqa: BLE001 - fail open on cache errors + logger.warning( + "Feature flag cache read failed for %s: %s", cache_key, e + ) + return default + + # 2) Resolve from the DB. + try: + rows = self._fetch_rows(flag_key, community_id) + except Exception as e: # noqa: BLE001 - fail open on DB errors + logger.warning( + "Feature flag DB lookup failed for %s: %s", flag_key, e + ) + return default + + row = self._pick_most_specific(rows, community_id, platform) + if row is None: + # Absent flag -> feature on (fail open). Not cached: the flag may be + # created at any moment and we want to pick it up promptly. + return default + + if not row.is_enabled: + result = False + elif row.rollout_pct >= 100: + result = True + elif row.rollout_pct <= 0: + result = False + else: + result = self._bucket_enabled(flag_key, community_id, row.rollout_pct) + + # 3) Cache the resolved boolean (best-effort). + try: + if self.redis is not None: + await self.redis.setex( + cache_key, self.default_ttl, b"1" if result else b"0" + ) + except Exception as e: # noqa: BLE001 - caching is best effort + logger.warning( + "Feature flag cache write failed for %s: %s", cache_key, e + ) + + return result + + async def publish_reload( + self, flag_key: str, community_id: int | None = None + ) -> None: + """ + Broadcast a cache-invalidation reload for ``flag_key`` on the + ``feature_flags:reload`` channel. Best-effort: logs and swallows errors. + """ + payload = json.dumps({"flag_key": flag_key, "community_id": community_id}) + try: + if self.redis is not None: + await self.redis.publish(RELOAD_CHANNEL, payload) + except Exception as e: # noqa: BLE001 - reload publish is best effort + logger.warning( + "Feature flag reload publish failed for %s: %s", flag_key, e + ) + + async def handle_reload(self, message: dict[str, Any]) -> None: + """ + Handle a reload message (a payload dict, or a raw Redis Pub/Sub message + whose ``data`` holds the JSON payload) by invalidating every cached key + for the referenced flag (``feature_flag:*:{flag_key}:*``). + """ + try: + payload = self._extract_payload(message) + flag_key = payload.get("flag_key") + if not flag_key: + return + await self._invalidate(flag_key) + except Exception as e: # noqa: BLE001 - invalidation is best effort + logger.warning("Feature flag reload handling failed: %s", e) + + # ------------------------------------------------------------------ # + # Internals + # ------------------------------------------------------------------ # + def _fetch_rows(self, flag_key: str, community_id: int | None) -> list[_FlagRow]: + """Fetch all candidate rows for ``flag_key`` scoped to the community.""" + result = self.dal.executesql( + """SELECT community_id, platform, is_enabled, rollout_pct + FROM feature_flags + WHERE flag_key = %s + AND (community_id IS NULL OR community_id = %s)""", + [flag_key, community_id], + ) + rows: list[_FlagRow] = [] + for raw in result or []: + rows.append( + _FlagRow( + community_id=raw[0], + platform=raw[1], + is_enabled=bool(raw[2]), + rollout_pct=int(raw[3]) if raw[3] is not None else 100, + ) + ) + return rows + + @staticmethod + def _pick_most_specific( + rows: list[_FlagRow], community_id: int | None, platform: str | None + ) -> _FlagRow | None: + """ + Return the highest-specificity row matching the scope, or ``None``. + + Community specificity dominates platform specificity, so a + community-scoped row always beats a global row even when the global row + is platform-specific. + """ + best: _FlagRow | None = None + best_score = -1 + for row in rows: + comm_specific = row.community_id is not None and row.community_id == community_id + comm_global = row.community_id is None + if not (comm_specific or comm_global): + continue + + plat_specific = ( + row.platform is not None + and platform is not None + and row.platform == platform + ) + plat_global = row.platform is None + if not (plat_specific or plat_global): + continue + + # Community rank weighted above platform rank so it always dominates. + score = (2 if comm_specific else 1) * 10 + (2 if plat_specific else 1) + if score > best_score: + best_score = score + best = row + return best + + async def _invalidate(self, flag_key: str) -> None: + """SCAN + DEL every cache key for ``flag_key`` (raw-client path).""" + if self.redis is None: + return + pattern = self._invalidate_pattern(flag_key) + async for key in self.redis.scan_iter(match=pattern): + await self.redis.delete(key) + + @staticmethod + def _decode_bool(cached: Any) -> bool: + """Coerce a cached Redis value (bytes or str, ``1``/``0``) to bool.""" + if isinstance(cached, (bytes, bytearray)): + cached = cached.decode() + return str(cached) == "1" + + @staticmethod + def _extract_payload(message: dict[str, Any]) -> dict[str, Any]: + """ + Normalize a reload message into its payload dict. Accepts both a bare + payload ``{"flag_key": ...}`` and a Redis Pub/Sub envelope whose + ``data`` holds the JSON string. + """ + data = message.get("data") if isinstance(message, dict) else None + if data is not None and not isinstance(data, dict): + if isinstance(data, (bytes, bytearray)): + data = data.decode() + try: + parsed = json.loads(data) + if isinstance(parsed, dict): + return parsed + except (ValueError, TypeError): + pass + return message + + +def create_feature_flag_service( + redis_client: Any, dal: Any, default_ttl: int = 300 +) -> FeatureFlagService: + """Factory for :class:`FeatureFlagService`.""" + return FeatureFlagService(redis_client=redis_client, dal=dal, default_ttl=default_ttl) diff --git a/services/core-community/libs/flask_core/flask_core/__init__.py b/services/core-community/libs/flask_core/flask_core/__init__.py index 4e6264c0..234057ef 100644 --- a/services/core-community/libs/flask_core/flask_core/__init__.py +++ b/services/core-community/libs/flask_core/flask_core/__init__.py @@ -36,6 +36,7 @@ record_request_metrics ) from .cache import CacheManager, create_cache_manager +from .feature_flags import FeatureFlagService, create_feature_flag_service from .rate_limiter import RateLimiter, RateLimitExceeded, create_rate_limiter from .message_queue import MessageQueue, Message, create_message_queue from .stream_pipeline import StreamPipeline, StreamEvent, create_stream_pipeline @@ -148,6 +149,9 @@ # Cache "CacheManager", "create_cache_manager", + # Feature Flags + "FeatureFlagService", + "create_feature_flag_service", # Rate Limiting "RateLimiter", "RateLimitExceeded", diff --git a/services/core-community/libs/flask_core/flask_core/feature_flags.py b/services/core-community/libs/flask_core/flask_core/feature_flags.py new file mode 100644 index 00000000..a2d434b9 --- /dev/null +++ b/services/core-community/libs/flask_core/flask_core/feature_flags.py @@ -0,0 +1,313 @@ +""" +Feature-Flag Evaluation Service +================================ + +Shared, Redis-cached feature-flag evaluator for all Waddles modules. + +Backing table (migration 068):: + + feature_flags(id, flag_key, community_id NULL, platform NULL, + is_enabled, rollout_pct, description, updated_by, + created_at, updated_at) + UNIQUE (flag_key, COALESCE(community_id, -1), COALESCE(platform, '*')) + +Resolution picks the *most specific* matching row for a +``(flag_key, community_id, platform)`` lookup, in this order (first match +wins):: + + 1. (community_id, platform) -- exact community + exact platform + 2. (community_id, NULL) -- exact community, all platforms + 3. (NULL, platform) -- global community, exact platform + 4. (NULL, NULL) -- global default + +Fail-open semantics: when no row matches, or when Redis / the DB raise, the +caller-supplied ``default`` is returned (an absent flag means the feature is +on). A matched row with ``is_enabled = false`` is an explicit kill-switch and +returns ``False``. When ``is_enabled = true`` and ``rollout_pct < 100`` a +deterministic, sticky bucketing decides the result so the same +``(flag_key, community_id)`` always lands the same way. + +Redis-client vs CacheManager +---------------------------- +This service takes a **raw** ``redis.asyncio`` client (the same object the +Router already holds and the object ``AiChatterConfigCache`` consumes), not a +``flask_core.CacheManager``. Rationale: the Router and the platform receivers +pass their bare Redis connection around directly, and this service needs +``PUBLISH`` + ``SCAN`` for reload fan-out / cache invalidation -- surface that +``CacheManager`` does not expose. A raw client is therefore the lowest common +denominator that *both* the Router and other ``flask_core`` consumers already +have on hand, so no adapter is required at any call site. The ``__init__`` +signature stays ``(redis_client, dal, default_ttl)`` regardless. +""" + +from __future__ import annotations + +import hashlib +import json +import logging +from dataclasses import dataclass +from typing import Any + +logger = logging.getLogger(__name__) + +#: Redis Pub/Sub channel used to broadcast cache-invalidation reloads. +RELOAD_CHANNEL = "feature_flags:reload" + + +@dataclass(slots=True) +class _FlagRow: + """A single ``feature_flags`` row materialized for specificity ranking.""" + + community_id: int | None + platform: str | None + is_enabled: bool + rollout_pct: int + + +class FeatureFlagService: + """ + Evaluate feature flags with a most-specific-wins resolution and a + Redis-cached final boolean. + + All public evaluation paths are fail-open: any Redis or DB error is logged + at ``warning`` level and the caller-supplied ``default`` is returned. + """ + + def __init__(self, redis_client: Any, dal: Any, default_ttl: int = 300) -> None: + """ + Args: + redis_client: Raw ``redis.asyncio`` client (may be ``None`` -- the + service then falls straight through to the DB and never caches). + dal: Object exposing a synchronous ``executesql(sql, params)`` that + returns a list of row tuples (PyDAL ``AsyncDAL`` / DAL). + default_ttl: TTL in seconds for cached boolean results. + """ + self.redis = redis_client + self.dal = dal + self.default_ttl = default_ttl + + # ------------------------------------------------------------------ # + # Key helpers + # ------------------------------------------------------------------ # + @staticmethod + def _cache_key( + flag_key: str, community_id: int | None, platform: str | None + ) -> str: + scope = community_id if community_id is not None else "global" + return f"feature_flag:{scope}:{flag_key}:{platform or '*'}" + + @staticmethod + def _invalidate_pattern(flag_key: str) -> str: + # Matches every community/platform variant cached for this flag. + return f"feature_flag:*:{flag_key}:*" + + @staticmethod + def _bucket_enabled( + flag_key: str, community_id: int | None, rollout_pct: int + ) -> bool: + """Deterministic sticky bucketing: same inputs -> same decision.""" + scope = community_id if community_id is not None else "global" + digest = hashlib.sha256(f"{flag_key}:{scope}".encode()).hexdigest() + return int(digest, 16) % 100 < rollout_pct + + # ------------------------------------------------------------------ # + # Public API + # ------------------------------------------------------------------ # + async def is_enabled( + self, + flag_key: str, + community_id: int | None = None, + platform: str | None = None, + *, + default: bool = True, + ) -> bool: + """ + Resolve whether ``flag_key`` is enabled for the given scope. + + Checks the Redis cache first; on a miss, resolves from the DB (one + parameterized query fetching every row for ``flag_key`` scoped to the + community, then picks specificity in Python) and caches the boolean. + + Returns ``default`` when no row matches or on any Redis/DB failure. + """ + cache_key = self._cache_key(flag_key, community_id, platform) + + # 1) Cache fast-path. + try: + if self.redis is not None: + cached = await self.redis.get(cache_key) + if cached is not None: + return self._decode_bool(cached) + except Exception as e: # noqa: BLE001 - fail open on cache errors + logger.warning( + "Feature flag cache read failed for %s: %s", cache_key, e + ) + return default + + # 2) Resolve from the DB. + try: + rows = self._fetch_rows(flag_key, community_id) + except Exception as e: # noqa: BLE001 - fail open on DB errors + logger.warning( + "Feature flag DB lookup failed for %s: %s", flag_key, e + ) + return default + + row = self._pick_most_specific(rows, community_id, platform) + if row is None: + # Absent flag -> feature on (fail open). Not cached: the flag may be + # created at any moment and we want to pick it up promptly. + return default + + if not row.is_enabled: + result = False + elif row.rollout_pct >= 100: + result = True + elif row.rollout_pct <= 0: + result = False + else: + result = self._bucket_enabled(flag_key, community_id, row.rollout_pct) + + # 3) Cache the resolved boolean (best-effort). + try: + if self.redis is not None: + await self.redis.setex( + cache_key, self.default_ttl, b"1" if result else b"0" + ) + except Exception as e: # noqa: BLE001 - caching is best effort + logger.warning( + "Feature flag cache write failed for %s: %s", cache_key, e + ) + + return result + + async def publish_reload( + self, flag_key: str, community_id: int | None = None + ) -> None: + """ + Broadcast a cache-invalidation reload for ``flag_key`` on the + ``feature_flags:reload`` channel. Best-effort: logs and swallows errors. + """ + payload = json.dumps({"flag_key": flag_key, "community_id": community_id}) + try: + if self.redis is not None: + await self.redis.publish(RELOAD_CHANNEL, payload) + except Exception as e: # noqa: BLE001 - reload publish is best effort + logger.warning( + "Feature flag reload publish failed for %s: %s", flag_key, e + ) + + async def handle_reload(self, message: dict[str, Any]) -> None: + """ + Handle a reload message (a payload dict, or a raw Redis Pub/Sub message + whose ``data`` holds the JSON payload) by invalidating every cached key + for the referenced flag (``feature_flag:*:{flag_key}:*``). + """ + try: + payload = self._extract_payload(message) + flag_key = payload.get("flag_key") + if not flag_key: + return + await self._invalidate(flag_key) + except Exception as e: # noqa: BLE001 - invalidation is best effort + logger.warning("Feature flag reload handling failed: %s", e) + + # ------------------------------------------------------------------ # + # Internals + # ------------------------------------------------------------------ # + def _fetch_rows(self, flag_key: str, community_id: int | None) -> list[_FlagRow]: + """Fetch all candidate rows for ``flag_key`` scoped to the community.""" + result = self.dal.executesql( + """SELECT community_id, platform, is_enabled, rollout_pct + FROM feature_flags + WHERE flag_key = %s + AND (community_id IS NULL OR community_id = %s)""", + [flag_key, community_id], + ) + rows: list[_FlagRow] = [] + for raw in result or []: + rows.append( + _FlagRow( + community_id=raw[0], + platform=raw[1], + is_enabled=bool(raw[2]), + rollout_pct=int(raw[3]) if raw[3] is not None else 100, + ) + ) + return rows + + @staticmethod + def _pick_most_specific( + rows: list[_FlagRow], community_id: int | None, platform: str | None + ) -> _FlagRow | None: + """ + Return the highest-specificity row matching the scope, or ``None``. + + Community specificity dominates platform specificity, so a + community-scoped row always beats a global row even when the global row + is platform-specific. + """ + best: _FlagRow | None = None + best_score = -1 + for row in rows: + comm_specific = row.community_id is not None and row.community_id == community_id + comm_global = row.community_id is None + if not (comm_specific or comm_global): + continue + + plat_specific = ( + row.platform is not None + and platform is not None + and row.platform == platform + ) + plat_global = row.platform is None + if not (plat_specific or plat_global): + continue + + # Community rank weighted above platform rank so it always dominates. + score = (2 if comm_specific else 1) * 10 + (2 if plat_specific else 1) + if score > best_score: + best_score = score + best = row + return best + + async def _invalidate(self, flag_key: str) -> None: + """SCAN + DEL every cache key for ``flag_key`` (raw-client path).""" + if self.redis is None: + return + pattern = self._invalidate_pattern(flag_key) + async for key in self.redis.scan_iter(match=pattern): + await self.redis.delete(key) + + @staticmethod + def _decode_bool(cached: Any) -> bool: + """Coerce a cached Redis value (bytes or str, ``1``/``0``) to bool.""" + if isinstance(cached, (bytes, bytearray)): + cached = cached.decode() + return str(cached) == "1" + + @staticmethod + def _extract_payload(message: dict[str, Any]) -> dict[str, Any]: + """ + Normalize a reload message into its payload dict. Accepts both a bare + payload ``{"flag_key": ...}`` and a Redis Pub/Sub envelope whose + ``data`` holds the JSON string. + """ + data = message.get("data") if isinstance(message, dict) else None + if data is not None and not isinstance(data, dict): + if isinstance(data, (bytes, bytearray)): + data = data.decode() + try: + parsed = json.loads(data) + if isinstance(parsed, dict): + return parsed + except (ValueError, TypeError): + pass + return message + + +def create_feature_flag_service( + redis_client: Any, dal: Any, default_ttl: int = 300 +) -> FeatureFlagService: + """Factory for :class:`FeatureFlagService`.""" + return FeatureFlagService(redis_client=redis_client, dal=dal, default_ttl=default_ttl) diff --git a/tests/unit/test_feature_flags.py b/tests/unit/test_feature_flags.py new file mode 100644 index 00000000..9fb43e9a --- /dev/null +++ b/tests/unit/test_feature_flags.py @@ -0,0 +1,284 @@ +"""Unit tests for flask_core.feature_flags.FeatureFlagService. + +The service under test only depends on the standard library, so it is loaded +directly by file path to avoid importing the full ``flask_core`` package +(which pulls in pydal / redis / quart). Async coroutines are driven with +``asyncio.run`` so the suite needs no pytest-asyncio plugin or config. +""" +import asyncio +import fnmatch +import importlib.util +import json +import os +import sys + +import pytest + +# --- Load feature_flags.py standalone (stdlib-only module) ------------------ +_MODULE_PATH = os.path.join( + os.path.dirname(__file__), + "..", "..", "libs", "flask_core", "flask_core", "feature_flags.py", +) +_spec = importlib.util.spec_from_file_location("_ff_under_test", _MODULE_PATH) +assert _spec and _spec.loader +feature_flags = importlib.util.module_from_spec(_spec) +# Register before exec so @dataclass(slots=True) can resolve the module. +sys.modules[_spec.name] = feature_flags +_spec.loader.exec_module(feature_flags) + +FeatureFlagService = feature_flags.FeatureFlagService +create_feature_flag_service = feature_flags.create_feature_flag_service +RELOAD_CHANNEL = feature_flags.RELOAD_CHANNEL + + +def run(coro): + """Drive a coroutine to completion without pytest-asyncio.""" + return asyncio.run(coro) + + +# --- Fakes ------------------------------------------------------------------ +class FakeRedis: + """Minimal in-memory async stand-in for a raw redis.asyncio client.""" + + def __init__(self): + self.store: dict[str, bytes] = {} + self.published: list[tuple[str, str]] = [] + + async def get(self, key): + return self.store.get(key) + + async def setex(self, key, ttl, value): + if isinstance(value, str): + value = value.encode() + self.store[key] = value + + async def delete(self, key): + self.store.pop(key, None) + + async def publish(self, channel, message): + self.published.append((channel, message)) + + async def scan_iter(self, match="*"): + for key in list(self.store.keys()): + if fnmatch.fnmatch(key, match): + yield key + + +class StubDAL: + """Emulates PyDAL executesql for the feature_flags table. + + ``rows`` is a list of (community_id, platform, is_enabled, rollout_pct). + Emulates the ``community_id IS NULL OR community_id = %s`` filter so the + Python-side specificity resolution is exercised realistically. + """ + + def __init__(self, rows=None, raise_exc=None): + self.rows = rows or [] + self.raise_exc = raise_exc + self.calls = 0 + + def executesql(self, sql, params): + self.calls += 1 + if self.raise_exc is not None: + raise self.raise_exc + community_id = params[1] + return [ + r for r in self.rows + if r[0] is None or r[0] == community_id + ] + + +def make_service(rows=None, raise_exc=None, redis=None, ttl=300): + dal = StubDAL(rows=rows, raise_exc=raise_exc) + svc = create_feature_flag_service(redis, dal, default_ttl=ttl) + return svc, dal, redis + + +# --- Resolution specificity ------------------------------------------------- +def test_community_override_beats_global_killswitch(): + """MOST SPECIFIC wins: a community row overrides a global kill-switch.""" + rows = [ + (None, None, False, 100), # global kill-switch + (5, None, True, 100), # community 5 override -> on + ] + svc, _, _ = make_service(rows) + assert run(svc.is_enabled("flag", community_id=5)) is True + + +def test_global_killswitch_applies_to_other_communities(): + """A community without its own row falls back to the global kill-switch.""" + rows = [ + (None, None, False, 100), # global kill-switch + (5, None, True, 100), # only community 5 overrides + ] + svc, _, _ = make_service(rows) + assert run(svc.is_enabled("flag", community_id=99)) is False + + +def test_community_scope_dominates_platform_specificity(): + """Community specificity dominates: a community-wide row beats a + global platform-specific row even though the latter names the platform.""" + rows = [ + (None, "twitch", True, 100), # global, twitch-specific -> on + (5, None, False, 100), # community 5, all platforms -> off + ] + svc, _, _ = make_service(rows) + assert run(svc.is_enabled("flag", community_id=5, platform="twitch")) is False + + +def test_platform_specific_beats_platform_null_same_community(): + rows = [ + (5, None, True, 100), # community 5, all platforms + (5, "twitch", False, 100), # community 5, twitch -> most specific + ] + svc, _, _ = make_service(rows) + assert run(svc.is_enabled("flag", community_id=5, platform="twitch")) is False + # Different platform falls back to the community-wide row. + assert run(svc.is_enabled("flag", community_id=5, platform="discord")) is True + + +def test_global_platform_specific_selected_for_global_scope(): + rows = [ + (None, None, True, 100), + (None, "twitch", False, 100), + ] + svc, _, _ = make_service(rows) + assert run(svc.is_enabled("flag", community_id=None, platform="twitch")) is False + assert run(svc.is_enabled("flag", community_id=None, platform="discord")) is True + + +def test_matched_row_disabled_returns_false(): + svc, _, _ = make_service([(None, None, False, 100)]) + assert run(svc.is_enabled("flag")) is False + + +# --- Absent flag / defaults ------------------------------------------------- +def test_absent_flag_returns_default_true(): + svc, _, _ = make_service([]) + assert run(svc.is_enabled("missing")) is True + + +def test_absent_flag_respects_explicit_default_false(): + svc, _, _ = make_service([]) + assert run(svc.is_enabled("missing", default=False)) is False + + +# --- Rollout bucketing ------------------------------------------------------ +def test_rollout_100_always_true(): + svc, _, _ = make_service([(None, None, True, 100)]) + assert run(svc.is_enabled("flag", community_id=1)) is True + + +def test_rollout_0_always_false(): + svc, _, _ = make_service([(None, None, True, 0)]) + assert run(svc.is_enabled("flag", community_id=1)) is False + + +def test_rollout_bucketing_is_deterministic(): + """Same (flag_key, community_id) -> same decision. No cache involved.""" + svc, _, _ = make_service([(None, None, True, 50)], redis=None) + first = run(svc.is_enabled("flag", community_id=42)) + second = run(svc.is_enabled("flag", community_id=42)) + third = run(svc.is_enabled("flag", community_id=42)) + assert first == second == third + + +def test_rollout_distribution_roughly_matches_pct(): + """Over many community_ids, ~50% land enabled at rollout_pct=50.""" + svc, _, _ = make_service([(None, None, True, 50)], redis=None) + enabled = sum( + 1 for cid in range(2000) + if run(svc.is_enabled("dist_flag", community_id=cid)) + ) + ratio = enabled / 2000 + assert 0.42 < ratio < 0.58, f"ratio={ratio}" + + +# --- Error handling (fail-open) -------------------------------------------- +def test_db_error_returns_default(): + svc, _, _ = make_service(raise_exc=RuntimeError("db down")) + assert run(svc.is_enabled("flag", default=True)) is True + assert run(svc.is_enabled("flag", default=False)) is False + + +def test_cache_read_error_returns_default(): + class BrokenRedis(FakeRedis): + async def get(self, key): + raise RuntimeError("redis down") + + svc, _, _ = make_service([(None, None, True, 100)], redis=BrokenRedis()) + assert run(svc.is_enabled("flag", default=True)) is True + + +# --- Caching ---------------------------------------------------------------- +def test_cache_hit_skips_db(): + redis = FakeRedis() + key = FeatureFlagService._cache_key("flag", 5, "twitch") + redis.store[key] = b"0" + svc, dal, _ = make_service(raise_exc=RuntimeError("should not query"), redis=redis) + result = run(svc.is_enabled("flag", community_id=5, platform="twitch")) + assert result is False + assert dal.calls == 0 # DB never touched on a cache hit + + +def test_miss_populates_cache(): + redis = FakeRedis() + svc, dal, _ = make_service([(None, None, True, 100)], redis=redis) + assert run(svc.is_enabled("flag", community_id=7)) is True + assert dal.calls == 1 + key = FeatureFlagService._cache_key("flag", 7, None) + assert redis.store[key] == b"1" + # Second call is served from cache. + assert run(svc.is_enabled("flag", community_id=7)) is True + assert dal.calls == 1 + + +# --- Reload publish / handle ----------------------------------------------- +def test_publish_reload_emits_expected_payload(): + redis = FakeRedis() + svc, _, _ = make_service([], redis=redis) + run(svc.publish_reload("flag", community_id=5)) + assert len(redis.published) == 1 + channel, message = redis.published[0] + assert channel == RELOAD_CHANNEL + assert json.loads(message) == {"flag_key": "flag", "community_id": 5} + + +def test_handle_reload_invalidates_all_flag_keys(): + redis = FakeRedis() + redis.store[FeatureFlagService._cache_key("flag", 1, "twitch")] = b"1" + redis.store[FeatureFlagService._cache_key("flag", 2, None)] = b"0" + redis.store[FeatureFlagService._cache_key("flag", None, "discord")] = b"1" + redis.store[FeatureFlagService._cache_key("other", 1, None)] = b"1" # untouched + svc, _, _ = make_service([], redis=redis) + + run(svc.handle_reload({"flag_key": "flag", "community_id": None})) + + remaining = list(redis.store.keys()) + assert remaining == [FeatureFlagService._cache_key("other", 1, None)] + + +def test_handle_reload_accepts_pubsub_envelope(): + redis = FakeRedis() + redis.store[FeatureFlagService._cache_key("flag", 1, None)] = b"1" + svc, _, _ = make_service([], redis=redis) + + envelope = { + "type": "message", + "channel": RELOAD_CHANNEL, + "data": json.dumps({"flag_key": "flag", "community_id": None}), + } + run(svc.handle_reload(envelope)) + assert FeatureFlagService._cache_key("flag", 1, None) not in redis.store + + +def test_handle_reload_ignores_missing_flag_key(): + redis = FakeRedis() + redis.store[FeatureFlagService._cache_key("flag", 1, None)] = b"1" + svc, _, _ = make_service([], redis=redis) + run(svc.handle_reload({"community_id": 5})) # no flag_key -> no-op + assert FeatureFlagService._cache_key("flag", 1, None) in redis.store + + +if __name__ == "__main__": + raise SystemExit(pytest.main([__file__, "-q"])) From f85375df96acfca5944907d465b8a948a25008aa Mon Sep 17 00:00:00 2001 From: zero Date: Wed, 8 Jul 2026 10:33:09 -0400 Subject: [PATCH 4/5] Gate router dispatch behind feature flags New _dispatch_gate enforces core-module bypass (identity/workflow are never blockable, now enforced at dispatch rather than only in the admin API), the community module toggle, and a module. feature-flag check on both the command path and the interaction path - the latter previously dispatched with no enable check at all. Adds a background listener on feature_flags:reload for cache invalidation and threads platform into flag evaluation. 11 unit tests. Co-Authored-By: Claude Fable 5 --- processing/router_module/app.py | 63 +++- .../services/command_processor.py | 87 +++++- tests/unit/test_router_feature_flags.py | 287 ++++++++++++++++++ 3 files changed, 427 insertions(+), 10 deletions(-) create mode 100644 tests/unit/test_router_feature_flags.py diff --git a/processing/router_module/app.py b/processing/router_module/app.py index 725f104d..64845cce 100644 --- a/processing/router_module/app.py +++ b/processing/router_module/app.py @@ -11,6 +11,7 @@ init_database, setup_aaa_logging, StreamPipeline, + FeatureFlagService, ) from config import Config # noqa: E402 @@ -29,12 +30,13 @@ rate_limiter = None session_manager = None stream_pipeline = None +feature_flag_service = None stream_consumers_tasks = [] @app.before_serving async def startup(): - global dal, command_processor, cache_manager, rate_limiter, session_manager, stream_pipeline, stream_consumers_tasks + global dal, command_processor, cache_manager, rate_limiter, session_manager, stream_pipeline, feature_flag_service, stream_consumers_tasks from services.command_processor import CommandProcessor from services.cache_manager import CacheManager from services.rate_limiter import RateLimiter @@ -58,6 +60,9 @@ async def startup(): command_registry = CommandRegistry(dal, cache_manager) await command_registry.initialize() # Load initial commands from database ai_chatter_config_cache = AiChatterConfigCache(cache_manager.redis, dal) + # Feature-flag evaluator: shares the Router's raw Redis client (same object + # AiChatterConfigCache uses) so it can SCAN+DEL its own cache keys on reload. + feature_flag_service = FeatureFlagService(cache_manager.redis, dal) command_processor = CommandProcessor( dal, cache_manager, @@ -65,12 +70,14 @@ async def startup(): session_manager, command_registry, ai_chatter_config_cache=ai_chatter_config_cache, + feature_flag_service=feature_flag_service, ) app.config['command_processor'] = command_processor app.config['cache_manager'] = cache_manager app.config['rate_limiter'] = rate_limiter app.config['session_manager'] = session_manager + app.config['feature_flag_service'] = feature_flag_service # Start command reload listener (background task for marketplace module updates) task = asyncio.create_task(_command_reload_listener( @@ -80,6 +87,15 @@ async def startup(): stream_consumers_tasks.append(task) logger.system("Started command reload listener", action="command_reload_listener_start") + # Start feature-flag reload listener (invalidates cached flag decisions when + # the admin hub mutates a flag and publishes on 'feature_flags:reload'). + task = asyncio.create_task(_feature_flag_reload_listener( + feature_flag_service, + Config.REDIS_URL + )) + stream_consumers_tasks.append(task) + logger.system("Started feature flag reload listener", action="feature_flag_reload_listener_start") + # Initialize StreamPipeline if enabled if Config.STREAM_PIPELINE_ENABLED: stream_pipeline = StreamPipeline( @@ -223,6 +239,51 @@ async def _command_reload_listener(command_registry, redis_url): logger.error(f"Command reload listener error: {e}") +async def _feature_flag_reload_listener(feature_flag_service, redis_url): + """Listen for feature_flags:reload events from the admin hub. + + Subscribes to the Redis pub/sub channel 'feature_flags:reload' and asks the + FeatureFlagService to invalidate every cached decision for the mutated flag. + The admin hub publishes {"flag_key": ..., "community_id": ...|null} after + every flag mutation. handle_reload accepts the raw pub/sub envelope directly. + """ + # Try to use redis.asyncio (newer), fall back to aioredis (legacy) + aioredis_lib = None + try: + import redis.asyncio as aioredis_lib + except ImportError: + try: + import aioredis as aioredis_lib + except ImportError: + logger.warning("No async Redis library available for feature_flags:reload listener") + return + + try: + redis = await aioredis_lib.from_url(redis_url or 'redis://redis:6379') + pubsub = redis.pubsub() + await pubsub.subscribe('feature_flags:reload') + logger.system("Subscribed to feature_flags:reload Redis channel", action="pubsub_subscribe") + + async for message in pubsub.listen(): + if message['type'] == 'message': + try: + # handle_reload accepts a raw pub/sub envelope (extracts and + # JSON-decodes 'data') and SCAN+DELs the flag's cache keys. + await feature_flag_service.handle_reload(message) + logger.debug("Feature flag cache invalidated", data=message.get('data')) + except Exception as e: + logger.error(f"Error processing feature_flags:reload event: {e}") + except asyncio.CancelledError: + logger.system("Feature flag reload listener shutting down", action="feature_flag_reload_listener_stop") + try: + await pubsub.unsubscribe('feature_flags:reload') + await redis.close() + except Exception: + pass + except Exception as e: + logger.error(f"Feature flag reload listener error: {e}") + + @app.after_serving async def shutdown(): """Cleanup handler for graceful shutdown""" diff --git a/processing/router_module/services/command_processor.py b/processing/router_module/services/command_processor.py index 4be8d38d..a920e7ed 100644 --- a/processing/router_module/services/command_processor.py +++ b/processing/router_module/services/command_processor.py @@ -14,6 +14,14 @@ logger = logging.getLogger(__name__) +# Non-disableable core modules (docs/APP_STANDARDS.md; migration 051_fix_is_core_flags.sql). +# Only ``identity`` and ``workflow`` are ``is_core = TRUE`` — they can never be blocked by +# the module-enable toggle or the feature-flag gate. A module-level frozenset is used rather +# than a DB ``is_core`` lookup because the set is fixed by the standards/migration, is checked +# on the command hot-path, and the exact names are guaranteed by the ``commands.module_name`` +# column (LEFT JOIN hub_modules ON name). This keeps dispatch free of an extra cached query. +CORE_MODULE_NAMES: frozenset[str] = frozenset({"identity", "workflow"}) + class CommandProcessor: def __init__( @@ -25,6 +33,7 @@ def __init__( command_registry: CommandRegistry, stream_pipeline=None, ai_chatter_config_cache=None, + feature_flag_service=None, ): self.dal = dal self.cache = cache_manager @@ -36,6 +45,7 @@ def __init__( self._grpc_manager = get_grpc_manager() if Config.GRPC_ENABLED else None self.stream_pipeline = stream_pipeline # Optional Redis streams pipeline self.ai_chatter_config_cache = ai_chatter_config_cache # AIChatter config cache + self.feature_flag_service = feature_flag_service # Feature-flag evaluator (fail-open) # Context service: per-user community context overrides self.context_service = ContextService(dal, cache_manager) @@ -158,7 +168,10 @@ async def process_event(self, event_data: Dict[str, Any]) -> Dict[str, Any]: asyncio.create_task(self._record_reputation_event(cmd_event)) # Execute command - result = await self.execute_command(command, entity_id, user_id, message, session_id) + result = await self.execute_command( + command, entity_id, user_id, message, session_id, + platform=event_data.get('platform'), + ) # Check for workflows triggered by this command await self._check_and_trigger_workflows(command, entity_id, user_id, message, session_id, event_data) @@ -209,7 +222,7 @@ async def _process_slash_command( # Execute as regular command result = await self.execute_command( - command, entity_id, user_id, message, session_id + command, entity_id, user_id, message, session_id, platform=platform ) # Add interaction metadata for deferred responses @@ -255,6 +268,14 @@ async def _process_interaction( # Resolve community for this entity community_id = await self._get_community_for_entity(entity_id) + # Gate the interaction dispatch the same way execute_command does: the + # module enable-toggle + the feature-flag system, with core modules + # (identity, workflow) bypassing both. Historically the interaction path + # POSTed to the module with no enable check at all. + gate = await self._dispatch_gate(module_name, community_id, platform) + if gate is not None: + return {**gate, "session_id": session_id} + # Look up the module's HTTP URL from the command registry (fast) or DB (fallback) module_url: Optional[str] = None try: @@ -589,6 +610,53 @@ async def _record_reputation_event(self, event_data: Dict[str, Any]): except Exception as e: logger.warning(f"Failed to record reputation event: {e}") + async def _dispatch_gate( + self, + module_name: str, + community_id: Optional[int], + platform: Optional[str] = None, + ) -> Optional[Dict[str, Any]]: + """Gate a module dispatch behind the enable-toggle and feature-flag system. + + Returns an error-response dict (without ``session_id`` — the caller adds + it) when the module must be blocked, or ``None`` when dispatch may proceed. + + Core modules (``identity``, ``workflow``) are never blocked: per + docs/APP_STANDARDS.md they are ``is_core = TRUE`` and bypass BOTH the + module-disabled toggle and the feature-flag gate. + + The feature-flag check is fail-open: an absent flag (or any Redis/DB + error inside the service) resolves to enabled. + """ + # Core modules can never be blocked. + if module_name in CORE_MODULE_NAMES: + return None + + # Callers use 'unknown' as the missing-platform sentinel; normalize to + # None so flag lookups cache under a single key per (flag, community). + if not platform or platform == 'unknown': + platform = None + + # 1) Community module enable/disable toggle. + if community_id and not await self._is_module_enabled(module_name, community_id): + return { + "success": False, + "error": f"The '{module_name}' module is disabled for this community", + } + + # 2) Feature-flag gate: flag key "module.", fail-open when absent. + if self.feature_flag_service is not None: + flag_key = f"module.{module_name}" + if not await self.feature_flag_service.is_enabled( + flag_key, community_id=community_id, platform=platform + ): + return { + "success": False, + "error": f"The '{module_name}' feature is currently disabled", + } + + return None + async def _is_module_enabled(self, module_name: str, community_id: int) -> bool: """Check if a module is enabled for a community. Cached in Redis.""" if not community_id: @@ -699,6 +767,7 @@ async def execute_command( user_id: str, message: str, session_id: str, + platform: Optional[str] = None, ) -> Dict[str, Any]: """Execute command asynchronously""" try: @@ -722,13 +791,13 @@ async def execute_command( "help_url": f"/commands" } - # Check if module is enabled for this community - if not await self._is_module_enabled(cmd_info.module_name, community_id): - return { - "success": False, - "error": f"The '{cmd_info.module_name}' module is disabled for this community", - "session_id": session_id - } + # Gate dispatch behind the module enable-toggle and feature-flag + # system. Core modules (identity, workflow) bypass both gates. + gate = await self._dispatch_gate( + cmd_info.module_name, community_id, platform + ) + if gate is not None: + return {**gate, "session_id": session_id} # Check if command is enabled if not cmd_info.is_enabled: diff --git a/tests/unit/test_router_feature_flags.py b/tests/unit/test_router_feature_flags.py new file mode 100644 index 00000000..706a3151 --- /dev/null +++ b/tests/unit/test_router_feature_flags.py @@ -0,0 +1,287 @@ +"""Unit tests for the Router's feature-flag / module-enable dispatch gate. + +The router's ``CommandProcessor`` normally pulls in aiohttp, quart, pydal and +the local ``config`` / ``services`` packages, none of which are installed in the +unit-test environment. Following the stdlib-only approach in +``test_feature_flags.py``, this module: + +* stubs the heavy top-level imports in ``sys.modules`` so ``command_processor`` + can be loaded by file path, and +* loads the real ``FeatureFlagService`` (also by file path), + +then exercises the real ``_dispatch_gate``, ``execute_command`` and +``_process_interaction`` coroutines bound to a lightweight fake ``self`` (the +class ``__init__`` — which builds a ContextService/grpc manager — is never run). + +Coverage: + * flag-disabled blocks dispatch (module HTTP call never made) + * flag absent allows dispatch (fail-open) + * core module (identity/workflow) bypasses BOTH the module toggle and the flag + * the interaction path is now gated (previously unchecked) +""" +import asyncio +import fnmatch +import importlib.util +import os +import sys +import types + +import pytest + +_HERE = os.path.dirname(__file__) +_ROUTER = os.path.join(_HERE, "..", "..", "processing", "router_module") +_FF_PATH = os.path.join( + _HERE, "..", "..", "libs", "flask_core", "flask_core", "feature_flags.py" +) + + +def run(coro): + """Drive a coroutine to completion without pytest-asyncio.""" + return asyncio.run(coro) + + +# --- Load the real FeatureFlagService standalone ---------------------------- +def _load_feature_flags(): + spec = importlib.util.spec_from_file_location("_ff_router_test", _FF_PATH) + assert spec and spec.loader + mod = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = mod + spec.loader.exec_module(mod) + return mod + + +feature_flags = _load_feature_flags() +FeatureFlagService = feature_flags.FeatureFlagService + + +# --- Load command_processor with the heavy deps stubbed out ----------------- +def _install_stub(name, **attrs): + mod = types.ModuleType(name) + for k, v in attrs.items(): + setattr(mod, k, v) + sys.modules[name] = mod + return mod + + +def _load_command_processor(): + if "_router_command_processor" in sys.modules: + return sys.modules["_router_command_processor"] + + # Minimal Config with only the attributes touched at import / call time. + class _Config: + GRPC_ENABLED = False + STREAM_PIPELINE_ENABLED = False + SERVICE_API_KEY = "" + ROUTER_REQUEST_TIMEOUT = 5 + ROUTER_ENTITY_CACHE_TTL = 300 + + # aiohttp is only referenced inside methods we stub on the fake self, but the + # module imports it at top level, so a bare stand-in is enough. + _install_stub("aiohttp") + _install_stub("config", Config=_Config) + + services_pkg = types.ModuleType("services") + services_pkg.__path__ = [] # mark as a package + sys.modules["services"] = services_pkg + + class _CommandInfo: # only used for typing / isinstance-free access + pass + + _install_stub( + "services.command_registry", + CommandRegistry=object, + CommandInfo=_CommandInfo, + ) + _install_stub("services.context_service", ContextService=object) + _install_stub("services.grpc_clients", get_grpc_manager=lambda: None) + _install_stub("services.ai_chatter_config_cache", AiChatterConfigCache=object) + + path = os.path.join(_ROUTER, "services", "command_processor.py") + spec = importlib.util.spec_from_file_location("_router_command_processor", path) + assert spec and spec.loader + mod = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = mod + spec.loader.exec_module(mod) + return mod + + +cp = _load_command_processor() +CommandProcessor = cp.CommandProcessor +CORE_MODULE_NAMES = cp.CORE_MODULE_NAMES + + +# --- Fakes ------------------------------------------------------------------ +class StubDAL: + """Emulates PyDAL executesql for the feature_flags table. + + ``rows`` is a list of (community_id, platform, is_enabled, rollout_pct). + """ + + def __init__(self, rows=None): + self.rows = rows or [] + + def executesql(self, sql, params=None): + # Feature-flag query: params == [flag_key, community_id] + if params and len(params) == 2 and "feature_flags" in sql: + community_id = params[1] + return [r for r in self.rows if r[0] is None or r[0] == community_id] + # Interaction module-URL lookup / anything else -> no rows. + return [] + + +def make_flag_service(rows=None): + return FeatureFlagService(redis_client=None, dal=StubDAL(rows=rows)) + + +class FakeSelf: + """A stand-in ``self`` carrying only what the gated coroutines touch.""" + + def __init__(self, *, flag_rows=None, community_id=5, module_enabled=True): + self.feature_flag_service = make_flag_service(rows=flag_rows) + self.dal = StubDAL() + self._community_id = community_id + self._module_enabled = module_enabled + self.module_calls = [] # (url, payload) for each dispatch attempt + self.handled_responses = [] + + async def _get_community_for_entity(self, entity_id, user_id=None, platform=None): + return self._community_id + + async def _is_module_enabled(self, module_name, community_id): + return self._module_enabled + + async def _call_module_with_retry(self, url, payload, max_retries=3): + self.module_calls.append((url, payload)) + return {"ok": True} + + async def handle_module_response(self, data): + self.handled_responses.append(data) + + # Bind the real gate/dispatch coroutines. + _dispatch_gate = CommandProcessor._dispatch_gate + execute_command = CommandProcessor.execute_command + _process_interaction = CommandProcessor._process_interaction + + +class Cmd: + """Lightweight CommandInfo stand-in.""" + + def __init__(self, module_name, module_url="http://mod:8000", + is_enabled=True, cooldown_seconds=0): + self.module_name = module_name + self.module_url = module_url + self.is_enabled = is_enabled + self.cooldown_seconds = cooldown_seconds + + +def with_command(fake, cmd): + """Attach a command_registry that returns ``cmd``.""" + class _Registry: + async def get_command(self, command, community_id): + return cmd + fake.command_registry = _Registry() + return fake + + +# --- _dispatch_gate directly ------------------------------------------------ +def test_gate_blocks_when_flag_disabled(): + fake = FakeSelf(flag_rows=[(None, None, False, 100)]) # global kill-switch + gate = run(fake._dispatch_gate("loyalty", 5)) + assert gate is not None + assert gate["success"] is False + assert "disabled" in gate["error"] + + +def test_gate_allows_when_flag_absent_fail_open(): + fake = FakeSelf(flag_rows=[]) # no flag rows -> fail open + assert run(fake._dispatch_gate("loyalty", 5)) is None + + +def test_gate_blocks_when_module_toggle_off(): + fake = FakeSelf(flag_rows=[], module_enabled=False) + gate = run(fake._dispatch_gate("loyalty", 5)) + assert gate is not None + assert "module is disabled" in gate["error"] + + +def test_gate_core_module_bypasses_both_gates(): + # Module toggle OFF *and* a global flag kill-switch -> core still passes. + for core in CORE_MODULE_NAMES: + fake = FakeSelf( + flag_rows=[(None, None, False, 100)], module_enabled=False + ) + assert run(fake._dispatch_gate(core, 5)) is None + assert CORE_MODULE_NAMES == frozenset({"identity", "workflow"}) + + +def test_gate_flag_is_platform_scoped(): + # Disabled only for twitch; discord stays enabled. + fake = FakeSelf(flag_rows=[(None, "twitch", False, 100)]) + assert run(fake._dispatch_gate("loyalty", 5, "twitch")) is not None + assert run(fake._dispatch_gate("loyalty", 5, "discord")) is None + + +# --- execute_command path --------------------------------------------------- +def test_execute_command_flag_disabled_blocks_dispatch(): + fake = with_command( + FakeSelf(flag_rows=[(None, None, False, 100)]), + Cmd("loyalty"), + ) + result = run(fake.execute_command("!points", "e1", "u1", "!points", "s1")) + assert result["success"] is False + assert "disabled" in result["error"] + assert result["session_id"] == "s1" + assert fake.module_calls == [] # module was never dispatched to + + +def test_execute_command_flag_absent_allows_dispatch(): + fake = with_command(FakeSelf(flag_rows=[]), Cmd("loyalty")) + result = run(fake.execute_command("!points", "e1", "u1", "!points", "s1")) + assert result["success"] is True + assert len(fake.module_calls) == 1 # module was dispatched to + + +def test_execute_command_core_module_bypasses_disabled_flag(): + fake = with_command( + FakeSelf(flag_rows=[(None, None, False, 100)], module_enabled=False), + Cmd("identity"), + ) + result = run(fake.execute_command("!whoami", "e1", "u1", "!whoami", "s1")) + assert result["success"] is True + assert len(fake.module_calls) == 1 # core module dispatched despite kill-switch + + +# --- interaction path (previously ungated) ---------------------------------- +def _interaction(fake, custom_id, platform="twitch"): + event_data = {"message_type": "button_click", "platform": platform} + metadata = {"custom_id": custom_id, "values": {}} + return run( + fake._process_interaction(event_data, "e1", "u1", "s1", metadata) + ) + + +def test_interaction_flag_disabled_blocks_dispatch(): + fake = FakeSelf(flag_rows=[(None, None, False, 100)]) + result = _interaction(fake, "loyalty:buy:item_1") + assert result["success"] is False + assert "disabled" in result["error"] + assert result["session_id"] == "s1" + assert fake.module_calls == [] # never POSTed to the module + + +def test_interaction_flag_absent_allows_dispatch(): + fake = FakeSelf(flag_rows=[]) + result = _interaction(fake, "loyalty:buy:item_1") + assert result["success"] is True + assert len(fake.module_calls) == 1 + + +def test_interaction_core_module_bypasses_gates(): + fake = FakeSelf(flag_rows=[(None, None, False, 100)], module_enabled=False) + result = _interaction(fake, "workflow:run:wf_9") + assert result["success"] is True + assert len(fake.module_calls) == 1 + + +if __name__ == "__main__": + raise SystemExit(pytest.main([__file__, "-q"])) From 8ae7e81cc7788a2150e69765099dceae02f11f5f Mon Sep 17 00:00:00 2001 From: zero Date: Wed, 8 Jul 2026 10:33:09 -0400 Subject: [PATCH 5/5] Add feature-flag admin surface to hub portal Community admins manage per-community overrides (scoped CRUD, cannot touch global or foreign rows); superadmins manage global flags and the audit trail. Effective-state display replicates the router's specificity resolution exactly (community scope outranks platform scope). Every mutation writes an audit row in-transaction and publishes feature_flags:reload via a new shared node-redis client. Redis ACLs grant the hub user publish and the router user subscribe on that channel - the router user previously had no channel permissions at all, which also silently broke its existing command:reload subscription. Co-Authored-By: Claude Fable 5 --- admin/hub_module/backend/package-lock.json | 101 ++++ admin/hub_module/backend/package.json | 1 + admin/hub_module/backend/src/config/redis.js | 118 +++++ .../controllers/featureFlagAdminController.js | 317 +++++++++++++ .../src/controllers/featureFlagController.js | 275 +++++++++++ .../backend/src/routes/featureFlags.js | 80 ++++ admin/hub_module/backend/src/routes/index.js | 4 + .../backend/src/routes/superadmin.js | 28 ++ .../src/services/featureFlagService.js | 297 ++++++++++++ admin/hub_module/frontend/src/App.jsx | 4 + .../frontend/src/layouts/AdminLayout.jsx | 3 + .../frontend/src/layouts/DashboardLayout.jsx | 2 + .../src/pages/admin/AdminFeatureFlags.jsx | 429 +++++++++++++++++ .../superadmin/SuperAdminFeatureFlags.jsx | 435 ++++++++++++++++++ admin/hub_module/frontend/src/services/api.js | 14 + config/redis/users.acl | 4 +- 16 files changed, 2110 insertions(+), 2 deletions(-) create mode 100644 admin/hub_module/backend/src/config/redis.js create mode 100644 admin/hub_module/backend/src/controllers/featureFlagAdminController.js create mode 100644 admin/hub_module/backend/src/controllers/featureFlagController.js create mode 100644 admin/hub_module/backend/src/routes/featureFlags.js create mode 100644 admin/hub_module/backend/src/services/featureFlagService.js create mode 100644 admin/hub_module/frontend/src/pages/admin/AdminFeatureFlags.jsx create mode 100644 admin/hub_module/frontend/src/pages/superadmin/SuperAdminFeatureFlags.jsx diff --git a/admin/hub_module/backend/package-lock.json b/admin/hub_module/backend/package-lock.json index fbc3425c..f4983892 100644 --- a/admin/hub_module/backend/package-lock.json +++ b/admin/hub_module/backend/package-lock.json @@ -23,6 +23,7 @@ "multer": "^2.0.2", "nodemailer": "8.0.5", "pg": "^8.13.1", + "redis": "^4.7.0", "socket.io": "^4.8.1", "uuid": "^14.0.0", "xss": "^1.0.15" @@ -825,6 +826,65 @@ "node": ">=20.0.0" } }, + "node_modules/@redis/bloom": { + "version": "1.2.0", + "resolved": "https://registry.npmjs.org/@redis/bloom/-/bloom-1.2.0.tgz", + "integrity": "sha512-HG2DFjYKbpNmVXsa0keLHp/3leGJz1mjh09f2RLGGLQZzSHpkmZWuwJbAvo3QcRY8p80m5+ZdXZdYOSBLlp7Cg==", + "license": "MIT", + "peerDependencies": { + "@redis/client": "^1.0.0" + } + }, + "node_modules/@redis/client": { + "version": "1.6.1", + "resolved": "https://registry.npmjs.org/@redis/client/-/client-1.6.1.tgz", + "integrity": "sha512-/KCsg3xSlR+nCK8/8ZYSknYxvXHwubJrU82F3Lm1Fp6789VQ0/3RJKfsmRXjqfaTA++23CvC3hqmqe/2GEt6Kw==", + "license": "MIT", + "dependencies": { + "cluster-key-slot": "1.1.2", + "generic-pool": "3.9.0", + "yallist": "4.0.0" + }, + "engines": { + "node": ">=14" + } + }, + "node_modules/@redis/graph": { + "version": "1.1.1", + "resolved": "https://registry.npmjs.org/@redis/graph/-/graph-1.1.1.tgz", + "integrity": "sha512-FEMTcTHZozZciLRl6GiiIB4zGm5z5F3F6a6FZCyrfxdKOhFlGkiAqlexWMBzCi4DcRoyiOsuLfW+cjlGWyExOw==", + "license": "MIT", + "peerDependencies": { + "@redis/client": "^1.0.0" + } + }, + "node_modules/@redis/json": { + "version": "1.0.7", + "resolved": "https://registry.npmjs.org/@redis/json/-/json-1.0.7.tgz", + "integrity": "sha512-6UyXfjVaTBTJtKNG4/9Z8PSpKE6XgSyEb8iwaqDcy+uKrd/DGYHTWkUdnQDyzm727V7p21WUMhsqz5oy65kPcQ==", + "license": "MIT", + "peerDependencies": { + "@redis/client": "^1.0.0" + } + }, + "node_modules/@redis/search": { + "version": "1.2.0", + "resolved": "https://registry.npmjs.org/@redis/search/-/search-1.2.0.tgz", + "integrity": "sha512-tYoDBbtqOVigEDMAcTGsRlMycIIjwMCgD8eR2t0NANeQmgK/lvxNAvYyb6bZDD4frHRhIHkJu2TBRvB0ERkOmw==", + "license": "MIT", + "peerDependencies": { + "@redis/client": "^1.0.0" + } + }, + "node_modules/@redis/time-series": { + "version": "1.1.0", + "resolved": "https://registry.npmjs.org/@redis/time-series/-/time-series-1.1.0.tgz", + "integrity": "sha512-c1Q99M5ljsIuc4YdaCwfUEXsofakb9c8+Zse2qxTadu8TalLXuAESzLvFAvNVbkmSlvlzIQOLpBCmWI9wTOt+g==", + "license": "MIT", + "peerDependencies": { + "@redis/client": "^1.0.0" + } + }, "node_modules/@simplewebauthn/server": { "version": "13.2.3", "resolved": "https://registry.npmjs.org/@simplewebauthn/server/-/server-13.2.3.tgz", @@ -1290,6 +1350,15 @@ "url": "https://github.com/chalk/chalk?sponsor=1" } }, + "node_modules/cluster-key-slot": { + "version": "1.1.2", + "resolved": "https://registry.npmjs.org/cluster-key-slot/-/cluster-key-slot-1.1.2.tgz", + "integrity": "sha512-RMr0FhtfXemyinomL4hrWcYJxmX6deFdCxpJzhDttxgO1+bcCnkk+9drydLVDmAMG7NE6aN/fl4F7ucU/90gAA==", + "license": "Apache-2.0", + "engines": { + "node": ">=0.10.0" + } + }, "node_modules/color-convert": { "version": "2.0.1", "resolved": "https://registry.npmjs.org/color-convert/-/color-convert-2.0.1.tgz", @@ -2055,6 +2124,15 @@ "url": "https://github.com/sponsors/ljharb" } }, + "node_modules/generic-pool": { + "version": "3.9.0", + "resolved": "https://registry.npmjs.org/generic-pool/-/generic-pool-3.9.0.tgz", + "integrity": "sha512-hymDOu5B53XvN4QT9dBmZxPX4CWhBPPLguTZ9MMFeFa/Kg0xWVfylOVNlJji/E7yTZWFd/q9GO5TxDLq156D7g==", + "license": "MIT", + "engines": { + "node": ">= 4" + } + }, "node_modules/get-intrinsic": { "version": "1.3.0", "resolved": "https://registry.npmjs.org/get-intrinsic/-/get-intrinsic-1.3.0.tgz", @@ -3027,6 +3105,23 @@ "node": ">= 6" } }, + "node_modules/redis": { + "version": "4.7.1", + "resolved": "https://registry.npmjs.org/redis/-/redis-4.7.1.tgz", + "integrity": "sha512-S1bJDnqLftzHXHP8JsT5II/CtHWQrASX5K96REjWjlmWKrviSOLWmM7QnRLstAWsu1VBBV1ffV6DzCvxNP0UJQ==", + "license": "MIT", + "workspaces": [ + "./packages/*" + ], + "dependencies": { + "@redis/bloom": "1.2.0", + "@redis/client": "1.6.1", + "@redis/graph": "1.1.1", + "@redis/json": "1.0.7", + "@redis/search": "1.2.0", + "@redis/time-series": "1.1.0" + } + }, "node_modules/reflect-metadata": { "version": "0.2.2", "resolved": "https://registry.npmjs.org/reflect-metadata/-/reflect-metadata-0.2.2.tgz", @@ -3617,6 +3712,12 @@ "node": ">=0.4" } }, + "node_modules/yallist": { + "version": "4.0.0", + "resolved": "https://registry.npmjs.org/yallist/-/yallist-4.0.0.tgz", + "integrity": "sha512-3wdGidZyq5PB084XLES5TpOSRA3wjXAlIWMhum2kRcv/41Sn2emQ0dycQW4uZXLejwKvg6EsvbdlVL+FYEct7A==", + "license": "ISC" + }, "node_modules/yocto-queue": { "version": "0.1.0", "resolved": "https://registry.npmjs.org/yocto-queue/-/yocto-queue-0.1.0.tgz", diff --git a/admin/hub_module/backend/package.json b/admin/hub_module/backend/package.json index bdec2df2..b5b50f7d 100644 --- a/admin/hub_module/backend/package.json +++ b/admin/hub_module/backend/package.json @@ -25,6 +25,7 @@ "multer": "^2.0.2", "nodemailer": "8.0.5", "pg": "^8.13.1", + "redis": "^4.7.0", "socket.io": "^4.8.1", "uuid": "^14.0.0", "xss": "^1.0.15", diff --git a/admin/hub_module/backend/src/config/redis.js b/admin/hub_module/backend/src/config/redis.js new file mode 100644 index 00000000..dc46c249 --- /dev/null +++ b/admin/hub_module/backend/src/config/redis.js @@ -0,0 +1,118 @@ +/** + * Redis Configuration + * A single lazily-connected shared client using node-redis (redis v4). + * + * The hub backend uses Redis only for lightweight pub/sub cache-invalidation + * signalling (e.g. the "feature_flags:reload" channel the Python router + * subscribes to). It is NOT a hard dependency: if REDIS_URL is unset the + * exported client is null, and if Redis is unreachable every operation fails + * soft (logged, never thrown). The app must run fine with Redis down or absent. + */ +import { createClient } from 'redis'; +import { logger } from '../utils/logger.js'; + +// REDIS_URL example (see docker-compose.yml): redis://hub:@infra-redis:6379/0 +const redisUrl = process.env.REDIS_URL || null; + +/** + * Build the shared client. Connection is deferred until first use so that + * importing this module never blocks startup or crashes when Redis is absent. + * Returns null when no REDIS_URL is configured. + */ +function buildClient() { + if (!redisUrl) { + logger.debug('Redis disabled: REDIS_URL is not set'); + return null; + } + + const c = createClient({ + url: redisUrl, + socket: { + connectTimeout: 5000, + // Give up after a handful of attempts so an absent Redis does not + // produce an endless reconnect/error loop in the logs. + reconnectStrategy: (retries) => (retries > 10 ? false : Math.min(retries * 200, 3000)), + }, + }); + + // An 'error' listener is mandatory: without one, node-redis emits on the + // process and an unhandled 'error' would crash the app. + c.on('error', (err) => logger.warn('Redis client error', { error: err.message })); + c.on('ready', () => logger.info('Redis client ready')); + c.on('end', () => logger.debug('Redis connection closed')); + + return c; +} + +// Single shared client instance (or null when Redis is not configured). +const client = buildClient(); + +// De-dupe concurrent connect attempts. +let connecting = null; + +/** + * Return a connected client, or null if Redis is unavailable/unconfigured. + * Never throws. + * @returns {Promise} + */ +export async function getRedisClient() { + if (!client) return null; + if (client.isOpen) return client; + + if (!connecting) { + connecting = client.connect().catch((err) => { + logger.warn('Redis connection failed', { error: err.message }); + return null; + }).finally(() => { + connecting = null; + }); + } + await connecting; + return client.isOpen ? client : null; +} + +/** + * Fire-and-forget publish. Returns true if the message was handed to Redis, + * false if Redis is unavailable. Never throws. + * @param {string} channel + * @param {string} message + * @returns {Promise} + */ +export async function publish(channel, message) { + const c = await getRedisClient(); + if (!c) return false; + await c.publish(channel, message); + return true; +} + +/** + * Check Redis connectivity. Never throws. + * @returns {Promise} + */ +export async function checkConnection() { + try { + const c = await getRedisClient(); + if (!c) return false; + const pong = await c.ping(); + return pong === 'PONG'; + } catch { + return false; + } +} + +/** + * Close the shared client (best-effort, for graceful shutdown). + */ +export async function closeRedis() { + if (client && client.isOpen) { + try { + await client.quit(); + logger.info('Redis client closed'); + } catch (err) { + logger.warn('Error closing Redis client', { error: err.message }); + } + } +} + +export { client }; +export default { getRedisClient, publish, checkConnection, closeRedis, client }; diff --git a/admin/hub_module/backend/src/controllers/featureFlagAdminController.js b/admin/hub_module/backend/src/controllers/featureFlagAdminController.js new file mode 100644 index 00000000..1d010ffe --- /dev/null +++ b/admin/hub_module/backend/src/controllers/featureFlagAdminController.js @@ -0,0 +1,317 @@ +/** + * Feature Flag Admin Controller - superadmin surface. + * + * Superadmins manage GLOBAL flags (community_id NULL) and view the full audit + * trail. All mutations force community_id NULL, run in a transaction with the + * append-only feature_flag_audit INSERT, then publish a reload message. + * + * Kept in a dedicated controller (mounted from routes/superadmin.js) to match + * the existing pattern where superadmin.js aggregates several focused + * controllers (userManagementController, analyticsController, etc.). + */ +import { query, transaction } from '../config/database.js'; +import { errors } from '../middleware/errorHandler.js'; +import { logger } from '../utils/logger.js'; +import { + normalizeFlagKey, + normalizePlatform, + normalizeRolloutPct, + actorFromRequest, + insertFlagAudit, + publishReload, +} from '../services/featureFlagService.js'; + +const SCHEMA_ERROR_CODES = ['42P01', '42703', '42883']; +function isSchemaError(err) { + return SCHEMA_ERROR_CODES.includes(err?.code); +} + +function serializeFlag(row) { + return { + id: row.id, + flag_key: row.flag_key, + community_id: row.community_id, + platform: row.platform, + is_enabled: row.is_enabled, + rollout_pct: row.rollout_pct, + description: row.description, + updated_by: row.updated_by, + created_at: row.created_at?.toISOString?.() || row.created_at, + updated_at: row.updated_at?.toISOString?.() || row.updated_at, + }; +} + +/** + * GET /api/v1/superadmin/feature-flags + * List all GLOBAL flags with a count of community overrides per flag_key. + */ +export async function listGlobalFlags(req, res, next) { + try { + const search = req.query.search || ''; + const params = []; + let where = 'WHERE g.community_id IS NULL'; + if (search) { + params.push(`%${search}%`); + where += ` AND (g.flag_key ILIKE $${params.length} OR g.description ILIKE $${params.length})`; + } + + const result = await query( + `SELECT g.*, + (SELECT COUNT(*) FROM feature_flags o + WHERE o.flag_key = g.flag_key AND o.community_id IS NOT NULL) AS override_count + FROM feature_flags g + ${where} + ORDER BY g.flag_key, g.platform NULLS FIRST`, + params + ); + + const flags = result.rows.map((row) => ({ + ...serializeFlag(row), + override_count: parseInt(row.override_count || 0, 10), + })); + + res.json({ success: true, flags }); + } catch (err) { + if (isSchemaError(err)) { + return res.json({ success: true, flags: [] }); + } + next(err); + } +} + +/** + * POST /api/v1/superadmin/feature-flags + * Create a GLOBAL flag (community_id NULL). + */ +export async function createGlobalFlag(req, res, next) { + try { + const keyCheck = normalizeFlagKey(req.body.flag_key); + if (keyCheck.error) return next(errors.badRequest(keyCheck.error)); + const platCheck = normalizePlatform(req.body.platform); + if (platCheck.error) return next(errors.badRequest(platCheck.error)); + const pctCheck = normalizeRolloutPct(req.body.rollout_pct); + if (pctCheck.error) return next(errors.badRequest(pctCheck.error)); + + const { flagKey } = keyCheck; + const { platform } = platCheck; + const { rolloutPct } = pctCheck; + const isEnabled = req.body.is_enabled === undefined ? false : Boolean(req.body.is_enabled); + const description = typeof req.body.description === 'string' ? req.body.description : null; + const actor = actorFromRequest(req); + + const existing = await query( + `SELECT id FROM feature_flags + WHERE flag_key = $1 AND community_id IS NULL + AND COALESCE(platform, '*') = COALESCE($2, '*')`, + [flagKey, platform] + ); + if (existing.rows.length > 0) { + return next(errors.conflict('A global flag with this key/platform already exists')); + } + + const created = await transaction(async (client) => { + const result = await client.query( + `INSERT INTO feature_flags + (flag_key, community_id, platform, is_enabled, rollout_pct, description, updated_by) + VALUES ($1, NULL, $2, $3, $4, $5, $6) + RETURNING *`, + [flagKey, platform, isEnabled, rolloutPct, description, actor] + ); + const row = result.rows[0]; + await insertFlagAudit(client, { + flagKey, + communityId: null, + platform, + action: 'created', + oldValue: null, + newValue: serializeFlag(row), + changedBy: actor, + }); + return row; + }); + + await publishReload(flagKey, null); + logger.audit('Global feature flag created', { adminId: req.user?.id, flagKey, platform }); + + res.status(201).json({ success: true, flag: serializeFlag(created) }); + } catch (err) { + next(err); + } +} + +/** + * PUT /api/v1/superadmin/feature-flags/:id + * Update a GLOBAL flag. The target row must be global (community_id NULL). + */ +export async function updateGlobalFlag(req, res, next) { + try { + const id = parseInt(req.params.id, 10); + if (isNaN(id)) return next(errors.badRequest('Invalid flag id')); + + const existing = await query('SELECT * FROM feature_flags WHERE id = $1', [id]); + if (existing.rows.length === 0) { + return next(errors.notFound('Feature flag not found')); + } + const current = existing.rows[0]; + if (current.community_id !== null) { + return next(errors.badRequest('This endpoint only manages global flags; use the community endpoint for overrides')); + } + + const updates = []; + const params = []; + let idx = 1; + + if (req.body.is_enabled !== undefined) { + updates.push(`is_enabled = $${idx++}`); + params.push(Boolean(req.body.is_enabled)); + } + if (req.body.rollout_pct !== undefined) { + const pctCheck = normalizeRolloutPct(req.body.rollout_pct); + if (pctCheck.error) return next(errors.badRequest(pctCheck.error)); + updates.push(`rollout_pct = $${idx++}`); + params.push(pctCheck.rolloutPct); + } + if (req.body.description !== undefined) { + updates.push(`description = $${idx++}`); + params.push(typeof req.body.description === 'string' ? req.body.description : null); + } + if (updates.length === 0) { + return next(errors.badRequest('No updates provided')); + } + + const actor = actorFromRequest(req); + updates.push(`updated_by = $${idx++}`); + params.push(actor); + updates.push('updated_at = NOW()'); + params.push(id); + + const updated = await transaction(async (client) => { + const result = await client.query( + `UPDATE feature_flags SET ${updates.join(', ')} WHERE id = $${idx} RETURNING *`, + params + ); + const row = result.rows[0]; + await insertFlagAudit(client, { + flagKey: row.flag_key, + communityId: null, + platform: row.platform, + action: 'updated', + oldValue: serializeFlag(current), + newValue: serializeFlag(row), + changedBy: actor, + }); + return row; + }); + + await publishReload(updated.flag_key, null); + logger.audit('Global feature flag updated', { adminId: req.user?.id, flagKey: updated.flag_key }); + + res.json({ success: true, flag: serializeFlag(updated) }); + } catch (err) { + next(err); + } +} + +/** + * DELETE /api/v1/superadmin/feature-flags/:id + * Delete a GLOBAL flag. The target row must be global (community_id NULL). + */ +export async function deleteGlobalFlag(req, res, next) { + try { + const id = parseInt(req.params.id, 10); + if (isNaN(id)) return next(errors.badRequest('Invalid flag id')); + + const existing = await query('SELECT * FROM feature_flags WHERE id = $1', [id]); + if (existing.rows.length === 0) { + return next(errors.notFound('Feature flag not found')); + } + const current = existing.rows[0]; + if (current.community_id !== null) { + return next(errors.badRequest('This endpoint only manages global flags')); + } + + const actor = actorFromRequest(req); + await transaction(async (client) => { + await client.query('DELETE FROM feature_flags WHERE id = $1', [id]); + await insertFlagAudit(client, { + flagKey: current.flag_key, + communityId: null, + platform: current.platform, + action: 'deleted', + oldValue: serializeFlag(current), + newValue: null, + changedBy: actor, + }); + }); + + await publishReload(current.flag_key, null); + logger.audit('Global feature flag deleted', { adminId: req.user?.id, flagKey: current.flag_key }); + + res.json({ success: true }); + } catch (err) { + next(err); + } +} + +/** + * GET /api/v1/superadmin/feature-flags/audit + * Paginated audit trail, optionally filtered by flag_key. + */ +export async function listAudit(req, res, next) { + try { + const page = Math.max(1, parseInt(req.query.page || '1', 10)); + const limit = Math.min(100, Math.max(1, parseInt(req.query.limit || '25', 10))); + const offset = (page - 1) * limit; + const flagKey = req.query.flag_key || ''; + + const params = []; + let where = 'WHERE 1=1'; + if (flagKey) { + params.push(flagKey); + where += ` AND flag_key = $${params.length}`; + } + + const countResult = await query(`SELECT COUNT(*) AS count FROM feature_flag_audit ${where}`, params); + const total = parseInt(countResult.rows[0]?.count || 0, 10); + + const result = await query( + `SELECT id, flag_key, community_id, platform, action, old_value, new_value, changed_by, changed_at + FROM feature_flag_audit + ${where} + ORDER BY changed_at DESC, id DESC + LIMIT $${params.length + 1} OFFSET $${params.length + 2}`, + [...params, limit, offset] + ); + + const entries = result.rows.map((row) => ({ + id: row.id, + flag_key: row.flag_key, + community_id: row.community_id, + platform: row.platform, + action: row.action, + old_value: row.old_value, + new_value: row.new_value, + changed_by: row.changed_by, + changed_at: row.changed_at?.toISOString?.() || row.changed_at, + })); + + res.json({ + success: true, + entries, + pagination: { page, limit, total, totalPages: Math.ceil(total / limit) }, + }); + } catch (err) { + if (isSchemaError(err)) { + return res.json({ success: true, entries: [], pagination: { page: 1, limit: 25, total: 0, totalPages: 0 } }); + } + next(err); + } +} + +export default { + listGlobalFlags, + createGlobalFlag, + updateGlobalFlag, + deleteGlobalFlag, + listAudit, +}; diff --git a/admin/hub_module/backend/src/controllers/featureFlagController.js b/admin/hub_module/backend/src/controllers/featureFlagController.js new file mode 100644 index 00000000..75219845 --- /dev/null +++ b/admin/hub_module/backend/src/controllers/featureFlagController.js @@ -0,0 +1,275 @@ +/** + * Feature Flag Controller - community admin surface. + * + * Community admins manage feature-flag OVERRIDES for their OWN community only. + * A community admin must NEVER be able to create/update/delete a global row + * (community_id NULL) or another community's row: community_id is always forced + * from the :communityId URL param and every mutation re-checks ownership. + * + * Every mutation is wrapped in a transaction together with the append-only + * feature_flag_audit INSERT, then publishes a cache-invalidation message. + */ +import { query, transaction } from '../config/database.js'; +import { errors } from '../middleware/errorHandler.js'; +import { logger } from '../utils/logger.js'; +import { + normalizeFlagKey, + normalizePlatform, + normalizeRolloutPct, + actorFromRequest, + resolveEffectiveFlags, + insertFlagAudit, + publishReload, +} from '../services/featureFlagService.js'; + +/** + * PostgreSQL error codes for missing schema objects. If migration 068 has not + * been applied yet, list endpoints degrade to empty data instead of a 500. + */ +const SCHEMA_ERROR_CODES = ['42P01', '42703', '42883']; +function isSchemaError(err) { + return SCHEMA_ERROR_CODES.includes(err?.code); +} + +/** Serialize a feature_flags row for API responses. */ +function serializeFlag(row) { + return { + id: row.id, + flag_key: row.flag_key, + community_id: row.community_id, + platform: row.platform, + is_enabled: row.is_enabled, + rollout_pct: row.rollout_pct, + description: row.description, + updated_by: row.updated_by, + created_at: row.created_at?.toISOString?.() || row.created_at, + updated_at: row.updated_at?.toISOString?.() || row.updated_at, + }; +} + +/** + * GET /api/v1/admin/:communityId/feature-flags + * Merged view: every global flag (community_id NULL) plus this community's + * overrides. Effective state per (flag_key, platform) row is resolved with the + * router's exact specificity ranking (see resolveEffectiveFlags / + * libs/flask_core/flask_core/feature_flags.py _pick_most_specific): + * (community, platform) > (community, NULL) > (NULL, platform) > (NULL, NULL) — + * so the page always shows what the bot actually does, e.g. a community + * all-platform override wins over a platform-specific global row. + */ +export async function listCommunityFlags(req, res, next) { + try { + const communityId = parseInt(req.params.communityId, 10); + + const globalsResult = await query( + `SELECT * FROM feature_flags + WHERE community_id IS NULL + ORDER BY flag_key, platform NULLS FIRST`, + [] + ); + const overridesResult = await query( + `SELECT * FROM feature_flags + WHERE community_id = $1 + ORDER BY flag_key, platform NULLS FIRST`, + [communityId] + ); + + const flags = resolveEffectiveFlags(globalsResult.rows, overridesResult.rows, communityId); + + res.json({ success: true, flags }); + } catch (err) { + if (isSchemaError(err)) { + return res.json({ success: true, flags: [] }); + } + next(err); + } +} + +/** + * POST /api/v1/admin/:communityId/feature-flags + * Create a community-scoped override. community_id is forced from the URL. + */ +export async function createCommunityOverride(req, res, next) { + try { + const communityId = parseInt(req.params.communityId, 10); + + const keyCheck = normalizeFlagKey(req.body.flag_key); + if (keyCheck.error) return next(errors.badRequest(keyCheck.error)); + const platCheck = normalizePlatform(req.body.platform); + if (platCheck.error) return next(errors.badRequest(platCheck.error)); + const pctCheck = normalizeRolloutPct(req.body.rollout_pct); + if (pctCheck.error) return next(errors.badRequest(pctCheck.error)); + + const { flagKey } = keyCheck; + const { platform } = platCheck; + const { rolloutPct } = pctCheck; + const isEnabled = req.body.is_enabled === undefined ? false : Boolean(req.body.is_enabled); + const description = typeof req.body.description === 'string' ? req.body.description : null; + const actor = actorFromRequest(req); + + // Reject duplicates for this community + platform (unique per + // flag_key, COALESCE(community_id,-1), COALESCE(platform,'*')). + const existing = await query( + `SELECT id FROM feature_flags + WHERE flag_key = $1 AND community_id = $2 + AND COALESCE(platform, '*') = COALESCE($3, '*')`, + [flagKey, communityId, platform] + ); + if (existing.rows.length > 0) { + return next(errors.conflict('An override for this flag/platform already exists for this community')); + } + + const created = await transaction(async (client) => { + const result = await client.query( + `INSERT INTO feature_flags + (flag_key, community_id, platform, is_enabled, rollout_pct, description, updated_by) + VALUES ($1, $2, $3, $4, $5, $6, $7) + RETURNING *`, + [flagKey, communityId, platform, isEnabled, rolloutPct, description, actor] + ); + const row = result.rows[0]; + await insertFlagAudit(client, { + flagKey, + communityId, + platform, + action: 'created', + oldValue: null, + newValue: serializeFlag(row), + changedBy: actor, + }); + return row; + }); + + await publishReload(flagKey, communityId); + logger.audit('Feature flag override created', { adminId: req.user?.id, communityId, flagKey, platform }); + + res.status(201).json({ success: true, flag: serializeFlag(created) }); + } catch (err) { + next(err); + } +} + +/** + * PUT /api/v1/admin/:communityId/feature-flags/:id + * Update a community-scoped override. Ownership is re-checked: the row must + * belong to THIS community (never a global row, never another community). + */ +export async function updateCommunityOverride(req, res, next) { + try { + const communityId = parseInt(req.params.communityId, 10); + const id = parseInt(req.params.id, 10); + if (isNaN(id)) return next(errors.badRequest('Invalid flag id')); + + const existing = await query('SELECT * FROM feature_flags WHERE id = $1', [id]); + if (existing.rows.length === 0) { + return next(errors.notFound('Feature flag override not found')); + } + const current = existing.rows[0]; + // Guard: a community admin may only touch their own community's overrides. + if (current.community_id !== communityId) { + return next(errors.forbidden('Cannot modify a flag outside this community')); + } + + const updates = []; + const params = []; + let idx = 1; + + if (req.body.is_enabled !== undefined) { + updates.push(`is_enabled = $${idx++}`); + params.push(Boolean(req.body.is_enabled)); + } + if (req.body.rollout_pct !== undefined) { + const pctCheck = normalizeRolloutPct(req.body.rollout_pct); + if (pctCheck.error) return next(errors.badRequest(pctCheck.error)); + updates.push(`rollout_pct = $${idx++}`); + params.push(pctCheck.rolloutPct); + } + if (req.body.description !== undefined) { + updates.push(`description = $${idx++}`); + params.push(typeof req.body.description === 'string' ? req.body.description : null); + } + if (updates.length === 0) { + return next(errors.badRequest('No updates provided')); + } + + const actor = actorFromRequest(req); + updates.push(`updated_by = $${idx++}`); + params.push(actor); + updates.push('updated_at = NOW()'); + params.push(id); + + const updated = await transaction(async (client) => { + const result = await client.query( + `UPDATE feature_flags SET ${updates.join(', ')} WHERE id = $${idx} RETURNING *`, + params + ); + const row = result.rows[0]; + await insertFlagAudit(client, { + flagKey: row.flag_key, + communityId, + platform: row.platform, + action: 'updated', + oldValue: serializeFlag(current), + newValue: serializeFlag(row), + changedBy: actor, + }); + return row; + }); + + await publishReload(updated.flag_key, communityId); + logger.audit('Feature flag override updated', { adminId: req.user?.id, communityId, flagKey: updated.flag_key }); + + res.json({ success: true, flag: serializeFlag(updated) }); + } catch (err) { + next(err); + } +} + +/** + * DELETE /api/v1/admin/:communityId/feature-flags/:id + * Remove a community-scoped override (reverts to the global default). + */ +export async function deleteCommunityOverride(req, res, next) { + try { + const communityId = parseInt(req.params.communityId, 10); + const id = parseInt(req.params.id, 10); + if (isNaN(id)) return next(errors.badRequest('Invalid flag id')); + + const existing = await query('SELECT * FROM feature_flags WHERE id = $1', [id]); + if (existing.rows.length === 0) { + return next(errors.notFound('Feature flag override not found')); + } + const current = existing.rows[0]; + if (current.community_id !== communityId) { + return next(errors.forbidden('Cannot delete a flag outside this community')); + } + + const actor = actorFromRequest(req); + await transaction(async (client) => { + await client.query('DELETE FROM feature_flags WHERE id = $1', [id]); + await insertFlagAudit(client, { + flagKey: current.flag_key, + communityId, + platform: current.platform, + action: 'deleted', + oldValue: serializeFlag(current), + newValue: null, + changedBy: actor, + }); + }); + + await publishReload(current.flag_key, communityId); + logger.audit('Feature flag override deleted', { adminId: req.user?.id, communityId, flagKey: current.flag_key }); + + res.json({ success: true }); + } catch (err) { + next(err); + } +} + +export default { + listCommunityFlags, + createCommunityOverride, + updateCommunityOverride, + deleteCommunityOverride, +}; diff --git a/admin/hub_module/backend/src/routes/featureFlags.js b/admin/hub_module/backend/src/routes/featureFlags.js new file mode 100644 index 00000000..c91af456 --- /dev/null +++ b/admin/hub_module/backend/src/routes/featureFlags.js @@ -0,0 +1,80 @@ +/** + * Feature Flag Routes - community admin access. + * + * Mounted under the same /admin prefix and protected exactly like the other + * community-admin routes (admin.js): requireAuth on the router, then + * requireCommunityAdmin per route (which enforces the :communityId scope). + * community_id is always taken from the URL param, never the body. + */ +import { Router } from 'express'; +import { body, param } from 'express-validator'; +import * as featureFlagController from '../controllers/featureFlagController.js'; +import { requireAuth, requireCommunityAdmin } from '../middleware/auth.js'; +import { validateRequest } from '../middleware/validation.js'; +import { PLATFORM_ALLOWLIST } from '../services/featureFlagService.js'; + +const router = Router(); + +router.use(requireAuth); + +// Validation chains for create/update override bodies. +const flagKeyValidator = body('flag_key') + .isString() + .bail() + .isLength({ min: 1, max: 100 }) + .withMessage('flag_key must be 1-100 characters') + .bail() + .matches(/^[a-z0-9_.-]+$/) + .withMessage('flag_key may only contain lowercase letters, digits, dot, dash and underscore'); + +const platformValidator = body('platform') + .optional({ nullable: true }) + .custom((v) => v === null || v === '' || v === 'all' || PLATFORM_ALLOWLIST.includes(v)) + .withMessage(`platform must be null/all or one of: ${PLATFORM_ALLOWLIST.join(', ')}`); + +const rolloutValidator = body('rollout_pct') + .optional() + .isInt({ min: 0, max: 100 }) + .withMessage('rollout_pct must be an integer between 0 and 100'); + +const enabledValidator = body('is_enabled').optional().isBoolean().withMessage('is_enabled must be a boolean'); +const descriptionValidator = body('description').optional({ nullable: true }).isString().isLength({ max: 5000 }); + +// List merged flag view for this community +router.get('/:communityId/feature-flags', requireCommunityAdmin, featureFlagController.listCommunityFlags); + +// Create a community-scoped override +router.post( + '/:communityId/feature-flags', + requireCommunityAdmin, + flagKeyValidator, + platformValidator, + rolloutValidator, + enabledValidator, + descriptionValidator, + validateRequest, + featureFlagController.createCommunityOverride +); + +// Update a community-scoped override +router.put( + '/:communityId/feature-flags/:id', + requireCommunityAdmin, + param('id').isInt({ min: 1 }), + rolloutValidator, + enabledValidator, + descriptionValidator, + validateRequest, + featureFlagController.updateCommunityOverride +); + +// Delete a community-scoped override +router.delete( + '/:communityId/feature-flags/:id', + requireCommunityAdmin, + param('id').isInt({ min: 1 }), + validateRequest, + featureFlagController.deleteCommunityOverride +); + +export default router; diff --git a/admin/hub_module/backend/src/routes/index.js b/admin/hub_module/backend/src/routes/index.js index e2dfef20..420a5459 100644 --- a/admin/hub_module/backend/src/routes/index.js +++ b/admin/hub_module/backend/src/routes/index.js @@ -31,6 +31,7 @@ import analyticsRoutes from './analytics.js'; import raffleCustomizationRoutes from './raffleCustomization.js'; import githubSyncRoutes from './githubSync.js'; import aiKnowledgeRoutes from './aiKnowledge.js'; +import featureFlagRoutes from './featureFlags.js'; const router = Router(); @@ -122,6 +123,9 @@ router.use('/', githubSyncRoutes); // AI knowledge base + ticket suggestion routes (admin) router.use('/admin', aiKnowledgeRoutes); +// Feature flag routes (community-admin — manage this community's flag overrides) +router.use('/admin', featureFlagRoutes); + // Interaction routes (hub channels, forums — admin + member) router.use('/admin', interactionAdminRoutes); router.use('/community', communityInteractionRouter); diff --git a/admin/hub_module/backend/src/routes/superadmin.js b/admin/hub_module/backend/src/routes/superadmin.js index 6d9f68db..1f8b8a99 100644 --- a/admin/hub_module/backend/src/routes/superadmin.js +++ b/admin/hub_module/backend/src/routes/superadmin.js @@ -6,8 +6,11 @@ import * as superadminController from '../controllers/superadminController.js'; import * as analyticsController from '../controllers/analyticsController.js'; import PlatformConfigController from '../controllers/platformConfigController.js'; import * as userManagementController from '../controllers/userManagementController.js'; +import * as featureFlagAdminController from '../controllers/featureFlagAdminController.js'; import { requireAuth, requireSuperAdmin } from '../middleware/auth.js'; import { validators, validationRules, validateRequest } from '../middleware/validation.js'; +import { body } from 'express-validator'; +import { PLATFORM_ALLOWLIST } from '../services/featureFlagService.js'; const router = Router(); @@ -130,6 +133,31 @@ router.post('/users/:userId/password-reset', userManagementController.generatePa router.post('/users/:userId/analytics-consumer-role', userManagementController.assignAnalyticsConsumerRole); router.get('/users/:userId/deletion-request', userManagementController.getUserDeletionRequest); +// Feature flag management (global flags + audit trail) +// NOTE: /audit is declared before /:id so it is not swallowed by the param route. +router.get('/feature-flags', featureFlagAdminController.listGlobalFlags); +router.get('/feature-flags/audit', featureFlagAdminController.listAudit); +router.post('/feature-flags', + body('flag_key').isString().bail().isLength({ min: 1, max: 100 }) + .matches(/^[a-z0-9_.-]+$/).withMessage('flag_key may only contain lowercase letters, digits, dot, dash and underscore'), + body('platform').optional({ nullable: true }) + .custom((v) => v === null || v === '' || v === 'all' || PLATFORM_ALLOWLIST.includes(v)) + .withMessage(`platform must be null/all or one of: ${PLATFORM_ALLOWLIST.join(', ')}`), + body('rollout_pct').optional().isInt({ min: 0, max: 100 }), + body('is_enabled').optional().isBoolean(), + body('description').optional({ nullable: true }).isString().isLength({ max: 5000 }), + validateRequest, + featureFlagAdminController.createGlobalFlag +); +router.put('/feature-flags/:id', + body('rollout_pct').optional().isInt({ min: 0, max: 100 }), + body('is_enabled').optional().isBoolean(), + body('description').optional({ nullable: true }).isString().isLength({ max: 5000 }), + validateRequest, + featureFlagAdminController.updateGlobalFlag +); +router.delete('/feature-flags/:id', featureFlagAdminController.deleteGlobalFlag); + // Tenant management router.get('/tenants', superadminController.listTenants); router.post('/tenants', superadminController.createTenant); diff --git a/admin/hub_module/backend/src/services/featureFlagService.js b/admin/hub_module/backend/src/services/featureFlagService.js new file mode 100644 index 00000000..f58bb6ac --- /dev/null +++ b/admin/hub_module/backend/src/services/featureFlagService.js @@ -0,0 +1,297 @@ +/** + * Feature Flag Service - shared helpers for the feature-flag admin surface. + * + * Covers three concerns shared by the community-scoped controller + * (featureFlagController.js) and the superadmin global controller + * (featureFlagAdminController.js): + * 1. Input validation constants (flag key format, platform allowlist). + * 2. Append-only audit-trail inserts (feature_flag_audit). + * 3. Cache-invalidation publish to the "feature_flags:reload" Redis channel. + */ +import { logger } from '../utils/logger.js'; +import { publish } from '../config/redis.js'; + +/** + * Allowed platform values for a flag scope. `null` (all platforms) is handled + * separately by the callers and is always valid. + * Mirrors the bot's supported platform set. + */ +export const PLATFORM_ALLOWLIST = [ + 'twitch', + 'discord', + 'slack', + 'youtube', + 'kick', + 'teams', + 'mattermost', + 'googlechat', +]; + +/** flag_key format per the schema contract: lowercase, digits, dot, dash, underscore. */ +export const FLAG_KEY_REGEX = /^[a-z0-9_.-]+$/; + +/** + * Normalize/validate a platform value coming from a request body. + * Returns { platform: string|null } on success or { error: string } on failure. + */ +export function normalizePlatform(value) { + if (value === undefined || value === null || value === '' || value === 'all') { + return { platform: null }; + } + if (typeof value !== 'string' || !PLATFORM_ALLOWLIST.includes(value)) { + return { error: `platform must be null/all or one of: ${PLATFORM_ALLOWLIST.join(', ')}` }; + } + return { platform: value }; +} + +/** + * Validate a rollout percentage. Returns { rolloutPct: number } or { error }. + * Accepts undefined → defaults to 100. + */ +export function normalizeRolloutPct(value) { + if (value === undefined || value === null || value === '') { + return { rolloutPct: 100 }; + } + const n = Number(value); + if (!Number.isInteger(n) || n < 0 || n > 100) { + return { error: 'rollout_pct must be an integer between 0 and 100' }; + } + return { rolloutPct: n }; +} + +/** Validate a flag_key. Returns { flagKey } or { error }. */ +export function normalizeFlagKey(value) { + if (typeof value !== 'string' || value.length === 0 || value.length > 100) { + return { error: 'flag_key is required and must be at most 100 characters' }; + } + if (!FLAG_KEY_REGEX.test(value)) { + return { error: 'flag_key may only contain lowercase letters, digits, dot, dash and underscore' }; + } + return { flagKey: value }; +} + +/** + * Build the "actor" string stored in updated_by / changed_by (VARCHAR 255). + */ +export function actorFromRequest(req) { + return req.user?.email || req.user?.username || (req.user?.id ? String(req.user.id) : 'unknown'); +} + +/** + * Pick the highest-specificity feature_flags row for a + * (community_id, platform) lookup. + * + * EXACT port of the router's resolution + * (libs/flask_core/flask_core/feature_flags.py, _pick_most_specific): + * community specificity dominates platform specificity, so a community-scoped + * row always beats a global row even when the global row is platform-specific. + * Ranking (first match wins): + * 1. (community_id, platform) score 22 + * 2. (community_id, NULL) score 21 + * 3. (NULL, platform) score 12 + * 4. (NULL, NULL) score 11 + * + * @param {Array} rows - feature_flags rows (community_id/platform may be null) + * @param {number|null} communityId - lookup community scope + * @param {string|null} platform - lookup platform scope + * @returns {Object|null} the winning row, or null when nothing matches + */ +export function pickMostSpecific(rows, communityId, platform) { + let best = null; + let bestScore = -1; + for (const row of rows) { + const commSpecific = row.community_id !== null && row.community_id === communityId; + const commGlobal = row.community_id === null; + if (!commSpecific && !commGlobal) continue; + + const platSpecific = row.platform !== null && platform !== null && row.platform === platform; + const platGlobal = row.platform === null; + if (!platSpecific && !platGlobal) continue; + + // Community rank weighted above platform rank so it always dominates. + const score = (commSpecific ? 2 : 1) * 10 + (platSpecific ? 2 : 1); + if (score > bestScore) { + bestScore = score; + best = row; + } + } + return best; +} + +/** Scope label for a winning row, e.g. 'community-all' or 'global-platform'. */ +function scopeOf(row) { + if (!row) return null; + const comm = row.community_id !== null ? 'community' : 'global'; + const plat = row.platform !== null ? 'platform' : 'all'; + return `${comm}-${plat}`; +} + +function toIso(value) { + return value?.toISOString?.() || value || null; +} + +/** + * Build the community admin's merged flag view. + * + * For each displayed (flag_key, platform) combination present in either the + * global rows or this community's overrides, the effective state is resolved + * with the router's exact specificity ranking (pickMostSpecific above): + * community override at that platform → community override at platform NULL + * → global at that platform → global at platform NULL. Both effective_enabled + * and effective_rollout_pct come from the single winning row, so the admin + * page always mirrors actual bot behavior (e.g. a community all-platform + * override beats a platform-specific global row). + * + * Pure function — no DB access — so it is directly unit-testable. + * + * @param {Array} globalRows - feature_flags rows with community_id NULL + * @param {Array} overrideRows - feature_flags rows for this community + * @param {number} communityId + * @returns {Array} display rows + */ +export function resolveEffectiveFlags(globalRows, overrideRows, communityId) { + const byKey = new Map(); + for (const row of [...globalRows, ...overrideRows]) { + if (!byKey.has(row.flag_key)) byKey.set(row.flag_key, []); + byKey.get(row.flag_key).push(row); + } + + const flags = []; + const sortedKeys = [...byKey.keys()].sort(); + for (const flagKey of sortedKeys) { + const rows = byKey.get(flagKey); + + // Display one row per distinct platform scope (null = all platforms first). + const platformSet = new Map(); + for (const row of rows) platformSet.set(row.platform ?? '*', row.platform ?? null); + const platforms = [...platformSet.values()].sort((a, b) => { + if (a === null) return -1; + if (b === null) return 1; + return a < b ? -1 : a > b ? 1 : 0; + }); + + for (const platform of platforms) { + // Winner across ALL candidate rows — mirrors the router's runtime lookup. + const winner = pickMostSpecific(rows, communityId, platform); + // What the router would resolve if this community had no overrides. + const globalWinner = pickMostSpecific( + rows.filter((r) => r.community_id === null), + null, + platform + ); + // Exact-key rows (used for edit/revert actions and metadata). + const exactOverride = rows.find( + (r) => r.community_id === communityId && (r.platform ?? null) === platform + ) || null; + const exactGlobal = rows.find( + (r) => r.community_id === null && (r.platform ?? null) === platform + ) || null; + + const winnerIsOverride = Boolean(winner && winner.community_id !== null); + const meta = exactOverride || exactGlobal || winner || {}; + + flags.push({ + flag_key: flagKey, + platform, + description: exactGlobal?.description ?? winner?.description ?? null, + // Which row actually won (matches bot behavior) — drives the badge. + is_override: winnerIsOverride, + winning_scope: scopeOf(winner), + winning_override_id: winnerIsOverride ? winner.id : null, + // Exact override at this (flag_key, platform), if any — the target for + // edit/revert actions (may differ from the winning row). + override_id: exactOverride ? exactOverride.id : null, + effective_enabled: winner ? winner.is_enabled : null, + effective_rollout_pct: winner ? winner.rollout_pct : null, + global_enabled: globalWinner ? globalWinner.is_enabled : null, + global_rollout_pct: globalWinner ? globalWinner.rollout_pct : null, + updated_by: meta.updated_by ?? null, + updated_at: toIso(meta.updated_at), + }); + } + } + return flags; +} + +/** + * Insert an append-only audit row. MUST be called inside the same transaction + * as the mutation it records (pass the transaction client). + * + * @param {import('pg').PoolClient} client - transaction client + * @param {Object} entry + * @param {string} entry.flagKey + * @param {number|null} entry.communityId + * @param {string|null} entry.platform + * @param {'created'|'updated'|'deleted'} entry.action + * @param {Object|null} entry.oldValue + * @param {Object|null} entry.newValue + * @param {string} entry.changedBy + */ +export async function insertFlagAudit(client, { flagKey, communityId, platform, action, oldValue, newValue, changedBy }) { + await client.query( + `INSERT INTO feature_flag_audit + (flag_key, community_id, platform, action, old_value, new_value, changed_by) + VALUES ($1, $2, $3, $4, $5, $6, $7)`, + [ + flagKey, + communityId ?? null, + platform ?? null, + action, + oldValue === undefined || oldValue === null ? null : JSON.stringify(oldValue), + newValue === undefined || newValue === null ? null : JSON.stringify(newValue), + changedBy, + ] + ); +} + +/** + * Publish a cache-invalidation message so the runtime services drop their + * cached copy of a flag. + * + * CACHE-INVALIDATION CONTRACT: publish JSON {"flag_key": "...", "community_id": } + * to the Redis channel "feature_flags:reload". The Python router subscribes to + * this channel and invalidates its cached flag decisions. + * + * Fire-and-forget: this never throws into the request path. If Redis is + * unavailable the failure is logged at warn level and the mutation still + * succeeds (runtime caches simply expire on their own TTL). + */ +export const FEATURE_FLAG_RELOAD_CHANNEL = 'feature_flags:reload'; + +export async function publishReload(flagKey, communityId) { + const payload = { flag_key: flagKey, community_id: communityId ?? null }; + try { + const delivered = await publish(FEATURE_FLAG_RELOAD_CHANNEL, JSON.stringify(payload)); + if (delivered) { + logger.debug('Published feature_flags reload', { + channel: FEATURE_FLAG_RELOAD_CHANNEL, + payload, + }); + } else { + logger.warn('feature_flags reload not published — Redis unavailable', { + channel: FEATURE_FLAG_RELOAD_CHANNEL, + payload, + }); + } + } catch (err) { + logger.warn('feature_flags reload publish failed', { + channel: FEATURE_FLAG_RELOAD_CHANNEL, + payload, + error: err.message, + }); + } +} + +export default { + PLATFORM_ALLOWLIST, + FLAG_KEY_REGEX, + FEATURE_FLAG_RELOAD_CHANNEL, + normalizePlatform, + normalizeRolloutPct, + normalizeFlagKey, + actorFromRequest, + pickMostSpecific, + resolveEffectiveFlags, + insertFlagAudit, + publishReload, +}; diff --git a/admin/hub_module/frontend/src/App.jsx b/admin/hub_module/frontend/src/App.jsx index 96d972e1..bd57cc9e 100644 --- a/admin/hub_module/frontend/src/App.jsx +++ b/admin/hub_module/frontend/src/App.jsx @@ -90,6 +90,7 @@ import AdminInventory from './pages/admin/AdminInventory'; import AdminRconServers from './pages/admin/AdminRconServers'; import AdminCommunityTokens from './pages/admin/AdminCommunityTokens'; import AdminCommands from './pages/admin/AdminCommands'; +import AdminFeatureFlags from './pages/admin/AdminFeatureFlags'; import AdminPlatformSettings from './pages/admin/AdminPlatformSettings'; import AdminLfgConfig from './pages/admin/AdminLfgConfig'; import AdminClipConfig from './pages/admin/AdminClipConfig'; @@ -136,6 +137,7 @@ import SuperAdminVendorRequests from './pages/superadmin/SuperAdminVendorRequest import SuperAdminUsers from './pages/superadmin/SuperAdminUsers'; import SuperAdminAnalytics from './pages/superadmin/SuperAdminAnalytics'; import SuperAdminTenants from './pages/superadmin/SuperAdminTenants'; +import SuperAdminFeatureFlags from './pages/superadmin/SuperAdminFeatureFlags'; // Tenant admin pages import TenantDashboard from './pages/tenant/TenantDashboard'; @@ -338,6 +340,7 @@ function App() { } /> } /> } /> + } /> } /> } /> } /> @@ -389,6 +392,7 @@ function App() { } /> } /> } /> + } /> } /> diff --git a/admin/hub_module/frontend/src/layouts/AdminLayout.jsx b/admin/hub_module/frontend/src/layouts/AdminLayout.jsx index 56b863e7..3fce7f53 100644 --- a/admin/hub_module/frontend/src/layouts/AdminLayout.jsx +++ b/admin/hub_module/frontend/src/layouts/AdminLayout.jsx @@ -41,6 +41,7 @@ import { ClipboardDocumentListIcon, UserPlusIcon, AcademicCapIcon, + FlagIcon, } from '@heroicons/react/24/outline'; function AdminLayout() { @@ -160,6 +161,7 @@ function AdminLayout() { { to: `/admin/${communityId}/translation`, icon: LanguageIcon, label: 'Translation' }, { to: `/admin/${communityId}/support`, icon: TicketIcon, label: 'Support Tickets' }, { to: `/admin/${communityId}/tokens`, icon: Cog6ToothIcon, label: 'Tokens' }, + { to: `/admin/${communityId}/feature-flags`, icon: FlagIcon, label: 'Feature Flags' }, { to: `/admin/${communityId}/inventory`, icon: InboxStackIcon, label: 'Inventory' }, ], }, @@ -197,6 +199,7 @@ function AdminLayout() { { to: '/superadmin/modules', icon: BuildingStorefrontIcon, label: 'Module Registry' }, { to: '/superadmin/analytics', icon: ChartBarIcon, label: 'Analytics' }, { to: '/superadmin/platform-config', icon: Cog6ToothIcon, label: 'Platform Config' }, + { to: '/superadmin/feature-flags', icon: FlagIcon, label: 'Feature Flags' }, { to: '/superadmin/tenants', icon: ServerStackIcon, label: 'Tenants' }, ]; diff --git a/admin/hub_module/frontend/src/layouts/DashboardLayout.jsx b/admin/hub_module/frontend/src/layouts/DashboardLayout.jsx index 6c30590c..234de2b2 100644 --- a/admin/hub_module/frontend/src/layouts/DashboardLayout.jsx +++ b/admin/hub_module/frontend/src/layouts/DashboardLayout.jsx @@ -20,6 +20,7 @@ import { TicketIcon, CubeIcon, CodeBracketIcon, + FlagIcon, } from '@heroicons/react/24/outline'; import { useMemo, useState } from 'react'; import GlobalBanner from '../components/GlobalBanner'; @@ -132,6 +133,7 @@ function DashboardLayout() { { name: 'Vendor Requests', href: '/superadmin/vendor-requests', icon: ShoppingCartIcon }, { name: 'Analytics', href: '/superadmin/analytics', icon: ChartBarIcon }, { name: 'Platform Config', href: '/superadmin/platform-config', icon: Cog6ToothIcon }, + { name: 'Feature Flags', href: '/superadmin/feature-flags', icon: FlagIcon }, ], }); } diff --git a/admin/hub_module/frontend/src/pages/admin/AdminFeatureFlags.jsx b/admin/hub_module/frontend/src/pages/admin/AdminFeatureFlags.jsx new file mode 100644 index 00000000..ad1791af --- /dev/null +++ b/admin/hub_module/frontend/src/pages/admin/AdminFeatureFlags.jsx @@ -0,0 +1,429 @@ +import { useState, useEffect } from 'react'; +import { useParams } from 'react-router-dom'; +import { + FlagIcon, + XMarkIcon, + CheckIcon, + ExclamationTriangleIcon, + PencilSquareIcon, + PlusIcon, + ArrowUturnLeftIcon, +} from '@heroicons/react/24/outline'; +import { adminApi } from '../../services/api'; + +const PLATFORM_OPTIONS = [ + { value: '', label: 'All platforms' }, + { value: 'twitch', label: 'Twitch' }, + { value: 'discord', label: 'Discord' }, + { value: 'slack', label: 'Slack' }, + { value: 'youtube', label: 'YouTube' }, + { value: 'kick', label: 'Kick' }, + { value: 'teams', label: 'Teams' }, + { value: 'mattermost', label: 'Mattermost' }, + { value: 'googlechat', label: 'Google Chat' }, +]; + +function AdminFeatureFlags() { + const { communityId } = useParams(); + const [flags, setFlags] = useState([]); + const [loading, setLoading] = useState(true); + const [error, setError] = useState(null); + const [message, setMessage] = useState(null); + const [busy, setBusy] = useState({}); + const [editModal, setEditModal] = useState(null); // { mode: 'create'|'edit', flag } + + useEffect(() => { + loadFlags(); + }, [communityId]); + + const loadFlags = async () => { + try { + setLoading(true); + setError(null); + const res = await adminApi.getFeatureFlags(communityId); + setFlags(res.data.flags || []); + } catch (err) { + setError(err.response?.data?.error?.message || 'Failed to load feature flags'); + } finally { + setLoading(false); + } + }; + + const rowKey = (f) => `${f.flag_key}::${f.platform || '*'}`; + + const toggleEffective = async (flag) => { + const key = rowKey(flag); + try { + setBusy((b) => ({ ...b, [key]: true })); + if (flag.override_id) { + // An override exists at exactly this (flag, platform) — update it. + await adminApi.updateFeatureFlagOverride(communityId, flag.override_id, { + is_enabled: !flag.effective_enabled, + }); + } else { + // No exact override here — the effective state comes from a broader + // community override or a global row. Create a platform-exact override, + // the most specific scope, so it always wins the router's resolution. + await adminApi.createFeatureFlagOverride(communityId, { + flag_key: flag.flag_key, + platform: flag.platform || null, + is_enabled: !flag.effective_enabled, + rollout_pct: flag.effective_rollout_pct ?? 100, + description: flag.description || null, + }); + } + setMessage({ type: 'success', text: 'Flag updated for this community' }); + loadFlags(); + } catch (err) { + setError(err.response?.data?.error?.message || 'Failed to update flag'); + } finally { + setBusy((b) => ({ ...b, [key]: false })); + } + }; + + const revertOverride = async (flag) => { + if (!flag.override_id) return; + if (!confirm('Remove this community override? The next most specific flag (a broader community override or the global default) will take effect.')) return; + const key = rowKey(flag); + try { + setBusy((b) => ({ ...b, [key]: true })); + await adminApi.deleteFeatureFlagOverride(communityId, flag.override_id); + setMessage({ type: 'success', text: 'Override removed — the next most specific flag now applies' }); + loadFlags(); + } catch (err) { + setError(err.response?.data?.error?.message || 'Failed to remove override'); + } finally { + setBusy((b) => ({ ...b, [key]: false })); + } + }; + + const platformLabel = (p) => PLATFORM_OPTIONS.find((o) => o.value === (p || ''))?.label || p; + + return ( +
+
+
+

Feature Flags

+

+ Override global feature flags for this community. Overrides only affect this community. +

+
+ +
+ + {error && ( +
+
+ + {error} +
+ +
+ )} + + {message && ( +
+
+ + {message.text} +
+ +
+ )} + + {loading ? ( +
+
+
+ ) : flags.length === 0 ? ( +
+ +

No Feature Flags

+

+ No global feature flags are defined yet. New overrides you create will appear here. +

+
+ ) : ( +
+ + + + + + + + + + + + + {flags.map((flag) => { + const key = rowKey(flag); + return ( + + + + + + + + + ); + })} + +
FlagPlatformEffective StateRollout %SourceActions
+
+
+ +
+
+

{flag.flag_key}

+ {flag.description && ( +

{flag.description}

+ )} +
+
+ 		+ + {platformLabel(flag.platform)} + + + + {flag.effective_rollout_pct}% + {flag.is_override ? ( + + {/* Reflect the row that actually won the router's resolution: + a platform row can be governed by the community's + all-platform override. */} + {flag.winning_scope === 'community-all' && flag.platform + ? 'Override (all platforms)' + : 'Override'} + + ) : ( + + Global default + + )} + + {flag.override_id ? ( + <> + + + + ) : ( + + )} +
+
+ )} + + {editModal && ( + setEditModal(null)} + onSaved={(text) => { + setEditModal(null); + setMessage({ type: 'success', text }); + loadFlags(); + }} + onError={(text) => setError(text)} + /> + )} +
+ ); +} + +function OverrideModal({ communityId, mode, flag, onClose, onSaved, onError }) { + // In "edit" mode the override already exists. In "create" mode from a global + // row, flag_key/platform are seeded and locked. From the "New Override" + // button (flag === null) all fields are editable. + const editing = mode === 'edit'; + const seeded = Boolean(flag); + const [form, setForm] = useState({ + flag_key: flag?.flag_key || '', + platform: flag?.platform || '', + is_enabled: flag ? Boolean(flag.effective_enabled) : false, + rollout_pct: flag?.effective_rollout_pct ?? 100, + description: flag?.description || '', + }); + const [saving, setSaving] = useState(false); + const [formError, setFormError] = useState(null); + + const submit = async (e) => { + e.preventDefault(); + setFormError(null); + + if (!editing) { + if (!/^[a-z0-9_.-]+$/.test(form.flag_key) || form.flag_key.length === 0 || form.flag_key.length > 100) { + setFormError('flag_key may only contain lowercase letters, digits, dot, dash and underscore (max 100 chars)'); + return; + } + } + const pct = Number(form.rollout_pct); + if (!Number.isInteger(pct) || pct < 0 || pct > 100) { + setFormError('rollout_pct must be an integer between 0 and 100'); + return; + } + + try { + setSaving(true); + if (editing) { + await adminApi.updateFeatureFlagOverride(communityId, flag.override_id, { + is_enabled: form.is_enabled, + rollout_pct: pct, + description: form.description || null, + }); + onSaved('Override updated'); + } else { + await adminApi.createFeatureFlagOverride(communityId, { + flag_key: form.flag_key, + platform: form.platform || null, + is_enabled: form.is_enabled, + rollout_pct: pct, + description: form.description || null, + }); + onSaved('Override created'); + } + } catch (err) { + const msg = err.response?.data?.error?.message + || err.response?.data?.error?.details?.[0]?.msg + || 'Failed to save override'; + setFormError(msg); + onError?.(msg); + } finally { + setSaving(false); + } + }; + + return ( +
+
+
+

+ {editing ? 'Edit Override' : 'Create Override'} +

+ +
+
+
+ {formError && ( +
+ {formError} +
+ )} +
+ + setForm({ ...form, flag_key: e.target.value })} + disabled={editing || seeded} + className="w-full px-3 py-2 bg-navy-800 border border-navy-700 rounded-lg text-sky-100 font-mono text-sm focus:outline-none focus:border-gold-500 disabled:opacity-60" + placeholder="my_feature.key" + /> +
+
+ + +
+
+ + setForm({ ...form, rollout_pct: e.target.value })} + className="w-full px-3 py-2 bg-navy-800 border border-navy-700 rounded-lg text-sky-100 focus:outline-none focus:border-gold-500" + /> +
+
+ +