feat(replay): viewed_by_me filter (#70967)

Follow up to #64924. We'll
support this search filter so users can filter by replays they have or
haven't viewed. Previously this was hard because: the viewed_by_id
column in Snuba is populated by Sentry user ids (from Postgres), which
cannot be looked up from the web app unless you're a superuser.

(Definitely my first time making this PR.)

src/sentry/replays/usecases/query/__init__.py

@@ -43,6 +43,48 @@
 from sentry.replays.usecases.query.fields import ComputedField, TagField
 from sentry.utils.snuba import raw_snql_query
 
+VIEWED_BY_ME_KEY_ALIASES = ["viewed_by_me", "seen_by_me"]
+NULL_VIEWED_BY_ID_VALUE = 0  # default value in clickhouse
+
+
+def handle_viewed_by_me_filters(
+    search_filters: Sequence[SearchFilter | str | ParenExpression], request_user_id: int | None
+) -> Sequence[SearchFilter | str | ParenExpression]:
+    """Translate "viewed_by_me" as it's not a valid Snuba field, but a convenience alias for the frontend"""
+    new_filters = []
+    for search_filter in search_filters:
+        if (
+            not isinstance(search_filter, SearchFilter)
+            or search_filter.key.name not in VIEWED_BY_ME_KEY_ALIASES
+        ):
+            new_filters.append(search_filter)
+            continue
+
+        # since the value is boolean, negations (!) are not supported
+        if search_filter.operator != "=":
+            raise ParseError(f"Invalid operator specified for `{search_filter.key.name}`")
+
+        value = search_filter.value.value
+        if not isinstance(value, str) or value.lower() not in ["true", "false"]:
+            raise ParseError(f"Could not parse value for `{search_filter.key.name}`")
+        value = value.lower() == "true"
+
+        if request_user_id is None:
+            # This case will only occur from programmer error.
+            # Note the replay index endpoint returns 401 automatically for unauthorized and anonymous users.
+            raise ValueError("Invalid user id")
+
+        operator = "=" if value else "!="
+        new_filters.append(
+            SearchFilter(
+                SearchKey("viewed_by_id"),
+                operator,
+                SearchValue(request_user_id),
+            )
+        )
+
+    return new_filters
+
 
 def handle_search_filters(
     search_config: dict[str, FieldProtocol],
@@ -163,6 +205,9 @@ def query_using_optimized_search(
             SearchFilter(SearchKey("environment"), "IN", SearchValue(environments)),
         ]
 
+    # Translate "viewed_by_me" filters, which are aliases for "viewed_by_id"
+    search_filters = handle_viewed_by_me_filters(search_filters, request_user_id)
+
     can_scalar_sort = sort_is_scalar_compatible(sort or "started_at")
     can_scalar_search = can_scalar_search_subquery(search_filters)
 

tests/sentry/replays/test_organization_replay_index.py

@@ -773,8 +773,24 @@ def test_get_replays_user_filters(self):
     def test_get_replays_user_filters_invalid_operator(self):
         self.create_project(teams=[self.team])
 
+        queries = [
+            "transaction.duration:>0",
+            "viewed_by_me:<true",
+            "seen_by_me:>false",
+            "!viewed_by_me:false",
+            "!seen_by_me:true",
+        ]
+
+        with self.feature(REPLAYS_FEATURES):
+            for query in queries:
+                response = self.client.get(self.url + f"?field=id&query={query}")
+                assert response.status_code == 400
+
+    def test_get_replays_user_filters_invalid_value(self):
+        self.create_project(teams=[self.team])
+
         with self.feature(REPLAYS_FEATURES):
-            response = self.client.get(self.url + "?field=id&query=transaction.duration:>0")
+            response = self.client.get(self.url + "?field=id&query=viewed_by_me:potato")
             assert response.status_code == 400
 
     def test_get_replays_user_sorts(self):