Preserve indexed event filtering

olymp/events.py

@@ -23,27 +23,6 @@
 _LOGGER = logging.getLogger(__name__)
 _SUBSCRIBERS: list[EventSubscriber] = []
 _SUBSCRIBERS_LOCK = threading.Lock()
-_LIST_EVENTS_QUERY = """
-    SELECT *
-    FROM events
-    WHERE (? IS NULL OR event_type = ?)
-      AND (? IS NULL OR run_id = ?)
-      AND (? IS NULL OR plan_id = ?)
-      AND (? IS NULL OR node_id = ?)
-    ORDER BY event_id ASC
-    LIMIT ?
-"""
-_LIST_EVENTS_AFTER_QUERY = """
-    SELECT *
-    FROM events
-    WHERE event_id > ?
-      AND (? IS NULL OR event_type = ?)
-      AND (? IS NULL OR run_id = ?)
-      AND (? IS NULL OR plan_id = ?)
-      AND (? IS NULL OR node_id = ?)
-    ORDER BY event_id ASC
-    LIMIT ?
-"""
 
 
 def subscribe_events(handler: EventSubscriber) -> Callable[[], None]:
@@ -149,23 +128,14 @@ def list(
         checked_run = _nullable_text(run_id, "run_id")
         checked_plan = _nullable_text(plan_id, "plan_id")
         checked_node = _nullable_text(node_id, "node_id")
-        filter_values: tuple[object, ...] = (
-            checked_type,
-            checked_type,
-            checked_run,
-            checked_run,
-            checked_plan,
-            checked_plan,
-            checked_node,
-            checked_node,
-            checked_limit,
+        query, values = _event_list_query(
+            after_event_id=checked_after,
+            limit=checked_limit,
+            event_type=checked_type,
+            run_id=checked_run,
+            plan_id=checked_plan,
+            node_id=checked_node,
         )
-        if checked_after is None:
-            query = _LIST_EVENTS_QUERY
-            values = filter_values
-        else:
-            query = _LIST_EVENTS_AFTER_QUERY
-            values = (checked_after, *filter_values)
         with closing(self._connect()) as db:
             rows = db.execute(query, values).fetchall()
         return [_event_record(row) for row in rows]
@@ -185,6 +155,36 @@ def emit_event(path: Path, event_type: str, **kwargs: Any) -> dict[str, Any]:
     return EventStore(path).append(event_type, **kwargs)
 
 
+def _event_list_query(
+    *,
+    after_event_id: int | None,
+    limit: int,
+    event_type: str | None,
+    run_id: str | None,
+    plan_id: str | None,
+    node_id: str | None,
+) -> tuple[str, tuple[object, ...]]:
+    clauses: list[str] = []
+    values: list[object] = []
+    if after_event_id is not None:
+        clauses.append("event_id > ?")
+        values.append(after_event_id)
+    for clause, value in (
+        ("event_type = ?", event_type),
+        ("run_id = ?", run_id),
+        ("plan_id = ?", plan_id),
+        ("node_id = ?", node_id),
+    ):
+        if value is not None:
+            clauses.append(clause)
+            values.append(value)
+    where = f"WHERE {' AND '.join(clauses)}" if clauses else ""
+    values.append(limit)
+    # Clause text comes only from the fixed literals above; every external value is bound.
+    query = "\n".join(("SELECT *", "FROM events", where, "ORDER BY event_id ASC", "LIMIT ?"))
+    return query, tuple(values)
+
+
 def _notify_subscribers(event: dict[str, Any]) -> None:
     with _SUBSCRIBERS_LOCK:
         subscribers = tuple(_SUBSCRIBERS)

tests/test_events.py

@@ -88,18 +88,42 @@ def test_event_store_filters_and_paginates(self) -> None:
                 limit=1,
             )
             injected_filter = store.list(event_type="plugin.test' OR 1=1 --")
+            cursor_query, cursor_values = events_module._event_list_query(
+                after_event_id=1,
+                limit=100,
+                event_type=None,
+                run_id=None,
+                plan_id=None,
+                node_id=None,
+            )
+            run_query, run_values = events_module._event_list_query(
+                after_event_id=None,
+                limit=100,
+                event_type=None,
+                run_id="run-a",
+                plan_id=None,
+                node_id=None,
+            )
             with sqlite3.connect(db_path) as db:
-                query_plan = db.execute(
-                    "EXPLAIN QUERY PLAN " + events_module._LIST_EVENTS_AFTER_QUERY,
-                    (1, None, None, None, None, None, None, None, None, 100),
+                cursor_plan = db.execute(
+                    "EXPLAIN QUERY PLAN " + cursor_query,
+                    cursor_values,
+                ).fetchall()
+                run_plan = db.execute(
+                    "EXPLAIN QUERY PLAN " + run_query,
+                    run_values,
                 ).fetchall()
 
         self.assertEqual([event["event_id"] for event in first_page], matching_ids[:1])
         self.assertEqual([event["event_id"] for event in second_page], matching_ids[1:])
         self.assertEqual(injected_filter, [])
         self.assertTrue(
-            any("SEARCH events USING INTEGER PRIMARY KEY" in str(row[3]) for row in query_plan),
-            query_plan,
+            any("SEARCH events USING INTEGER PRIMARY KEY" in str(row[3]) for row in cursor_plan),
+            cursor_plan,
+        )
+        self.assertTrue(
+            any("SEARCH events USING INDEX idx_events_run_id" in str(row[3]) for row in run_plan),
+            run_plan,
         )
 
     def test_subscriber_failure_is_isolated_and_secret_safe(self) -> None: