feat: expose distinct warehouse event names per connection

api/experimentation/dataclasses.py

@@ -24,6 +24,12 @@ class WarehouseEventStats:
     unique_events_count: int
 
 
+@dataclass(frozen=True)
+class WarehouseEventNames:
+    events: list[str]
+    is_truncated: bool
+
+
 @dataclass(frozen=True)
 class ExposureBucket:
     variant: str

api/experimentation/services.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import hashlib
+import json
 import time
 import typing
 from dataclasses import replace
@@ -42,6 +44,7 @@
     ResultsAggregates,
     ResultsSummary,
     RolloutSpec,
+    WarehouseEventNames,
     WarehouseEventStats,
 )
 from experimentation.metrics import (
@@ -102,9 +105,27 @@
 CLICKHOUSE_CONNECT_TIMEOUT_SECONDS = 5
 CLICKHOUSE_QUERY_TIMEOUT_SECONDS = 30
 CLICKHOUSE_VERIFY_TIMEOUT_SECONDS = 5
+CLICKHOUSE_EVENT_NAMES_TIMEOUT_SECONDS = 15
 CUSTOMER_EVENT_STATS_CACHE_SECONDS = 60
+EVENT_NAMES_CACHE_SECONDS = 300
+CUSTOMER_EVENT_NAMES_FAILURE_CACHE_SECONDS = 60
+WAREHOUSE_EVENT_NAMES_LIMIT = 500
+
+_CUSTOMER_EVENT_UNAVAILABLE = "unavailable"
+
+
+def _customer_cache_key(kind: str, connection: "WarehouseConnection") -> str:
+    """Key cached warehouse reads by the connection's non-secret details, so a
+    config or type change can neither serve nor store stale reads. Credentials
+    stay out of the key material: they don't determine what the warehouse
+    holds, so rotating them keeps the cache valid."""
+    details = json.dumps(
+        [connection.warehouse_type, connection.config],
+        sort_keys=True,
+    )
+    digest = hashlib.sha256(details.encode()).hexdigest()[:12]
+    return f"experimentation:customer_{kind}:{connection.id}:{digest}"
 
-_CUSTOMER_EVENT_STATS_UNAVAILABLE = "unavailable"
 
 # A delivery run stops taking on new objects after this long, leaving room for
 # the slowest possible in-flight insert to still land inside the task timeout.
@@ -142,16 +163,72 @@ def _get_clickhouse_client() -> Client:
     return Client(host, **kwargs)
 
 
-def get_unique_event_names(environment_key: str) -> list[str]:
-    """Return the distinct event names recorded for `environment_key`,
-    ordered alphabetically."""
-    rows = _get_clickhouse_client().execute(
-        "SELECT DISTINCT event FROM events "
-        "WHERE environment_key = %(environment_key)s "
-        "ORDER BY event",
-        {"environment_key": environment_key},
+_CLICKHOUSE_EVENT_NAMES_QUERY = (
+    "SELECT event FROM events "
+    "WHERE environment_key = %(environment_key)s "
+    "GROUP BY event ORDER BY max(timestamp) DESC LIMIT %(limit)s"
+)
+
+
+def _event_names_query_params(environment_key: str) -> dict[str, str | int]:
+    # Fetch one row past the limit so truncation is detectable.
+    return {
+        "environment_key": environment_key,
+        "limit": WAREHOUSE_EVENT_NAMES_LIMIT + 1,
+    }
+
+
+def _build_event_names(
+    rows: "Sequence[Sequence[typing.Any]]",
+) -> WarehouseEventNames:
+    names = [event for (event,) in rows]
+    return WarehouseEventNames(
+        events=names[:WAREHOUSE_EVENT_NAMES_LIMIT],
+        is_truncated=len(names) > WAREHOUSE_EVENT_NAMES_LIMIT,
     )
-    return [row[0] for row in rows]
+
+
+EVENT_NAMES_SUPPORTED_WAREHOUSE_TYPES = (
+    WarehouseType.FLAGSMITH,
+    WarehouseType.CLICKHOUSE,
+)
+
+
+def get_warehouse_event_names(
+    connection: "WarehouseConnection",
+    environment_key: str,
+) -> WarehouseEventNames | None:
+    if connection.warehouse_type == WarehouseType.CLICKHOUSE:
+        return _get_customer_clickhouse_event_names(connection, environment_key)
+    if connection.warehouse_type == WarehouseType.FLAGSMITH:
+        return _get_flagsmith_clickhouse_event_names(environment_key)
+    raise ValueError(f"Unsupported warehouse type: {connection.warehouse_type}")
+
+
+def _get_flagsmith_clickhouse_event_names(
+    environment_key: str,
+) -> WarehouseEventNames | None:
+    if not settings.EXPERIMENTATION_CLICKHOUSE_URL:
+        return None
+    cache_key = f"experimentation:event_names:{environment_key}"
+    cached = cache.get(cache_key)
+    if isinstance(cached, WarehouseEventNames):
+        return cached
+    try:
+        rows = _get_clickhouse_client().execute(
+            _CLICKHOUSE_EVENT_NAMES_QUERY,
+            _event_names_query_params(environment_key),
+        )
+    except Exception:
+        logger.warning(
+            "connection.event_names_failed",
+            environment__key=environment_key,
+            exc_info=True,
+        )
+        return None
+    event_names = _build_event_names(rows)
+    cache.set(cache_key, event_names, EVENT_NAMES_CACHE_SECONDS)
+    return event_names
 
 
 _EVENT_STATS_QUERY = (
@@ -1079,11 +1156,11 @@ def _get_customer_warehouse_event_stats_cached(
     ClickHouse instance, or None when it's unreachable. Results — including
     failures — are cached briefly so read endpoints don't open a connection to
     the customer's host on every request."""
-    cache_key = f"experimentation:customer_event_stats:{connection.id}"
+    cache_key = _customer_cache_key("event_stats", connection)
     cached = cache.get(cache_key)
     if isinstance(cached, WarehouseEventStats):
         return cached
-    if cached == _CUSTOMER_EVENT_STATS_UNAVAILABLE:
+    if cached == _CUSTOMER_EVENT_UNAVAILABLE:
         return None
     try:
         with warehouse_delivery_service.delivery_client(
@@ -1098,7 +1175,7 @@ def _get_customer_warehouse_event_stats_cached(
     except Exception:
         cache.set(
             cache_key,
-            _CUSTOMER_EVENT_STATS_UNAVAILABLE,
+            _CUSTOMER_EVENT_UNAVAILABLE,
             CUSTOMER_EVENT_STATS_CACHE_SECONDS,
         )
         logger.warning(
@@ -1109,3 +1186,41 @@ def _get_customer_warehouse_event_stats_cached(
         return None
     cache.set(cache_key, stats, CUSTOMER_EVENT_STATS_CACHE_SECONDS)
     return stats
+
+
+def _get_customer_clickhouse_event_names(
+    connection: "WarehouseConnection",
+    environment_key: str,
+) -> WarehouseEventNames | None:
+    """Query the customer's ClickHouse instance, caching results — including
+    failures — to spare their host repeated connections."""
+    cache_key = _customer_cache_key("event_names", connection)
+    cached = cache.get(cache_key)
+    if isinstance(cached, WarehouseEventNames):
+        return cached
+    if cached == _CUSTOMER_EVENT_UNAVAILABLE:
+        return None
+    try:
+        with warehouse_delivery_service.delivery_client(
+            connection,
+            send_receive_timeout=CLICKHOUSE_EVENT_NAMES_TIMEOUT_SECONDS,
+        ) as client:
+            rows = client.query(
+                _CLICKHOUSE_EVENT_NAMES_QUERY,
+                parameters=_event_names_query_params(environment_key),
+            ).result_rows
+    except Exception:
+        cache.set(
+            cache_key,
+            _CUSTOMER_EVENT_UNAVAILABLE,
+            CUSTOMER_EVENT_NAMES_FAILURE_CACHE_SECONDS,
+        )
+        logger.warning(
+            "connection.event_names_failed",
+            environment__id=connection.environment_id,
+            exc_info=True,
+        )
+        return None
+    event_names = _build_event_names(rows)
+    cache.set(cache_key, event_names, EVENT_NAMES_CACHE_SECONDS)
+    return event_names

api/experimentation/views.py

@@ -1,4 +1,5 @@
 import logging
+from dataclasses import asdict
 from datetime import timedelta
 from typing import Any
 
@@ -59,12 +60,14 @@
     WarehouseConnectionSerializer,
 )
 from experimentation.services import (
+    EVENT_NAMES_SUPPORTED_WAREHOUSE_TYPES,
     annotate_warehouse_event_stats,
     apply_experiment_rollout,
     create_experiment_audit_log,
     create_metric_audit_log,
     create_warehouse_audit_log,
     enable_experiment_rollout,
+    get_warehouse_event_names,
     mark_warehouse_pending_connection,
     refresh_warehouse_connection_status,
     transition_experiment_status,
@@ -106,7 +109,7 @@ def get_throttles(self) -> list[BaseThrottle]:
         ):
             self.throttle_scope = "warehouse_connection_write"
             return [*super().get_throttles(), ScopedRateThrottle()]
-        if self.action in ("list", "retrieve"):
+        if self.action in ("list", "retrieve", "events"):
             self.throttle_scope = "warehouse_connection_read"
             return [*super().get_throttles(), ScopedRateThrottle()]
         return super().get_throttles()
@@ -222,6 +225,45 @@ def test_warehouse_connection_config(
             {"status": connection.status, "status_detail": connection.status_detail}
         )
 
+    @extend_schema(
+        operation_id="api_v1_environments_warehouse_connections_events_list",
+        responses={
+            200: inline_serializer(
+                name="WarehouseEventNamesResult",
+                fields={
+                    "events": serializers.ListField(child=serializers.CharField()),
+                    "is_truncated": serializers.BooleanField(),
+                },
+            ),
+            400: inline_serializer(
+                name="WarehouseEventNamesUnsupported",
+                fields={"detail": serializers.CharField()},
+            ),
+            503: inline_serializer(
+                name="WarehouseEventNamesUnavailable",
+                fields={"detail": serializers.CharField()},
+            ),
+        },
+    )
+    @action(detail=True, methods=["get"], url_path="events")
+    def events(self, request: Request, **kwargs: object) -> Response:
+        """List the distinct event names in the connection's warehouse."""
+        connection: WarehouseConnection = self.get_object()
+        if connection.warehouse_type not in EVENT_NAMES_SUPPORTED_WAREHOUSE_TYPES:
+            return Response(
+                {"detail": "Event listing is not supported for this warehouse type."},
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+        event_names = get_warehouse_event_names(
+            connection, self.kwargs["environment_api_key"]
+        )
+        if event_names is None:
+            return Response(
+                {"detail": "The warehouse is currently unreachable."},
+                status=status.HTTP_503_SERVICE_UNAVAILABLE,
+            )
+        return Response(asdict(event_names))
+
     def create(self, request: Request, *args: object, **kwargs: object) -> Response:
         environment = self._get_environment()
         serializer = self.get_serializer(data=request.data)