✨(backend) implement feature flags in Posthog analytics backend

Implement feature flags related functions in
Posthog analytics backend.
We cache the results in django cache to avoid
too frequent calls to Posthog.
Especially for when we are checking multiple features in
the same user request.

src/backend/core/analytics/posthog.py

@@ -1,13 +1,16 @@
 """PostHog implementation of the analytics backend protocol."""
 
 import logging
-from typing import Any
+from typing import Any, Mapping
+
+from django.core.cache import cache
 
 from posthog import Posthog
 
 from ..models import User
-from . import AnalyticsBackend
+from .base import AnalyticsBackend
 from .events import AnalyticsEvent
+from .user_feature_flags import UserFeatureFlag
 
 logger = logging.getLogger(__name__)
 
@@ -20,6 +23,8 @@ class PostHogAnalytics(AnalyticsBackend):
         *,
         api_key: str,
         host: str = "https://eu.i.posthog.com",
+        feature_flags_cache_ttl: int = 60,
+        feature_flags_cache_prefix: str = "user_feature_flags:",
         **kwargs: Any,
     ) -> None:
 
@@ -30,6 +35,8 @@ class PostHogAnalytics(AnalyticsBackend):
             host=host,
             **kwargs,
         )
+        self._feature_flags_cache_ttl = feature_flags_cache_ttl
+        self._feature_flags_cache_prefix = feature_flags_cache_prefix
 
     @staticmethod
     def _distinct_id(user: User) -> str | None:
@@ -73,3 +80,37 @@ class PostHogAnalytics(AnalyticsBackend):
     def shutdown(self) -> None:
         """Flush pending events. Called on process exit."""
         self._client.shutdown()
+
+    def _fetch_user_feature_flags(
+        self, user: User
+    ) -> Mapping[UserFeatureFlag, bool | str | None]:
+        """Compute feature flags for a user."""
+
+        distinct_id = self._distinct_id(user)
+        if distinct_id is None:
+            return {}
+
+        flags = self._client.evaluate_flags(distinct_id)
+        out: dict[UserFeatureFlag, bool | str | None] = {}
+        for flag_key in UserFeatureFlag:
+            out[flag_key] = flags.get_flag(flag_key.value)
+
+        return out
+
+    def get_user_feature_flags(
+        self, user: User
+    ) -> Mapping[UserFeatureFlag, bool | str | None]:
+        """Get feature flags for a user. Caches the result for a short time."""
+        distinct_id = self._distinct_id(user)
+        if distinct_id is None:
+            return {}
+
+        try:
+            return cache.get_or_set(
+                f"{self._feature_flags_cache_prefix}{distinct_id}",
+                default=lambda: self._fetch_user_feature_flags(user),
+                timeout=self._feature_flags_cache_ttl,
+            )
+        except Exception:  # pylint: disable=broad-exception-caught
+            logger.exception("Failed to get feature flags for user %s", user.pk)
+            return {}

src/backend/core/tests/analytics/test_posthog.py

@@ -12,6 +12,7 @@ import pytest
 
 from core.analytics.events import AnalyticsEvent
 from core.analytics.posthog import PostHogAnalytics
+from core.analytics.user_feature_flags import UserFeatureFlag
 from core.factories import UserFactory
 
 pytestmark = pytest.mark.django_db
@@ -248,6 +249,60 @@ def test_capture_logs_the_failing_event_name_on_exception(mock_posthog_cls, capl
     assert any("PostHog capture failed" in record.message for record in caplog.records)
 
 
+# ==============================
+# feature flags
+# ==============================
+
+
+@patch("core.analytics.posthog.Posthog")
+def test_compute_feature_flags_returns_all_catalog_entries(mock_posthog_cls):
+    """Should map every declared feature flag key to the SDK evaluated value."""
+    backend = PostHogAnalytics(api_key="test-api-key")
+    user = UserFactory()
+
+    mock_posthog_cls.return_value.evaluate_flags.return_value.get_flag.return_value = (
+        True
+    )
+
+    flags = backend._fetch_user_feature_flags(user)
+
+    assert flags == {UserFeatureFlag.TRANSCRIPT_SUMMARY_ENABLED: True}
+    mock_posthog_cls.return_value.evaluate_flags.assert_called_once_with(str(user.pk))
+    mock_posthog_cls.return_value.evaluate_flags.return_value.get_flag.assert_called_once_with(
+        UserFeatureFlag.TRANSCRIPT_SUMMARY_ENABLED.value
+    )
+
+
+@patch("core.analytics.posthog.cache.get_or_set")
+@patch("core.analytics.posthog.Posthog")
+def test_get_feature_flags_uses_cache_get_or_set(
+    mock_posthog_cls, mock_cache_get_or_set
+):
+    """Should cache feature flags by user distinct id with configured TTL."""
+    cached_flags = {UserFeatureFlag.TRANSCRIPT_SUMMARY_ENABLED: False}
+    mock_cache_get_or_set.return_value = cached_flags
+    backend = PostHogAnalytics(api_key="test-api-key", feature_flags_cache_ttl=120)
+    user = UserFactory()
+
+    flags = backend.get_user_feature_flags(user)
+
+    assert flags == cached_flags
+    mock_cache_get_or_set.assert_called_once()
+    args, kwargs = mock_cache_get_or_set.call_args
+    assert kwargs["timeout"] == 120
+    assert args[0] == f"user_feature_flags:{user.pk}"
+    assert callable(kwargs["default"])
+
+
+@patch("core.analytics.posthog.Posthog")
+def test_get_feature_flags_returns_empty_dict_on_exception(mock_posthog_cls):
+    """Should swallow failures and return an empty mapping."""
+    backend = PostHogAnalytics(api_key="test-api-key")
+    user = UserFactory()
+    with patch("core.analytics.posthog.cache.get_or_set", side_effect=RuntimeError):
+        assert backend.get_user_feature_flags(user) == {}
+
+
 # ==============================
 # shutdown
 # ==============================