Feat/merge search

deployments/api/src/stitch/api/db/merge_candidate_actions.py

@@ -4,7 +4,7 @@
 from datetime import datetime, timezone
 from typing import Any
 
-from sqlalchemy import func, select
+from sqlalchemy import func, or_, select, union
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
 
@@ -20,6 +20,7 @@
     MergeCandidateCreateRequest,
     MergeCandidateDetailView,
     MergeCandidateListItemView,
+    OGFieldQueryParams,
     MergeCandidatePage,
     MergeCandidateQueryParams,
     MergeCandidateReviewRequest,
@@ -38,6 +39,7 @@
     ResourceModel,
 )
 from .og_field_resource_actions import apply_resource_merge
+from .queries import base_resource_query
 from .utils import coalesce_resources, coalesce_resources_with_sources
 
 
@@ -261,6 +263,45 @@ async def _load_candidate_model(
     return model
 
 
+# Postgres bigint; a longer digit string cannot be a resource id (and would
+# overflow the comparison).
+_MAX_RESOURCE_ID = 2**63 - 1
+
+
+async def _resource_ids_matching(
+    session: AsyncSession,
+    q: str,
+    licensed_sources: Collection[OGSISrcKey] | None,
+) -> list[int]:
+    """Ids of queue resources (candidate members and merged resources) that
+    match ``q``; see ``MergeCandidateFilterParams.q``."""
+    with named_query("merge_candidates.list.search_scope"):
+        queue_resource_ids = list(
+            await session.scalars(
+                union(
+                    select(MergeCandidateItemModel.resource_id),
+                    select(MergeCandidateModel.merged_resource_id).where(
+                        MergeCandidateModel.merged_resource_id.is_not(None)
+                    ),
+                )
+            )
+        )
+
+    # The resource list's own search, limited to the queue's resources. Sorting
+    # by id keeps it from pivoting a sort column it does not need.
+    text_match = base_resource_query(
+        OGFieldQueryParams(q=q, sort_by="id"),
+        licensed_sources,
+        resource_ids=queue_resource_ids,
+    )
+    with named_query("merge_candidates.list.search"):
+        matching = set(await session.scalars(text_match))
+
+    if q.isdigit() and int(q) <= _MAX_RESOURCE_ID and int(q) in queue_resource_ids:
+        matching.add(int(q))
+    return sorted(matching)
+
+
 async def list_merge_candidates(
     session: AsyncSession,
     params: MergeCandidateQueryParams,
@@ -270,6 +311,19 @@ async def list_merge_candidates(
     filtered = select(MergeCandidateModel.id)
     if params.status:
         filtered = filtered.where(MergeCandidateModel.status.in_(params.status))
+    q = (params.q or "").strip()
+    if q:
+        matching = await _resource_ids_matching(session, q, licensed_sources)
+        filtered = filtered.where(
+            or_(
+                MergeCandidateModel.merged_resource_id.in_(matching),
+                MergeCandidateModel.id.in_(
+                    select(MergeCandidateItemModel.merge_candidate_id).where(
+                        MergeCandidateItemModel.resource_id.in_(matching)
+                    )
+                ),
+            )
+        )
 
     with named_query("merge_candidates.list.count"):
         total = (

deployments/api/src/stitch/api/db/queries.py

@@ -388,36 +388,50 @@ def coalesced_candidate_rows(
     )
 
 
-def _resource_universe() -> Select[tuple[int]]:
+def _resource_universe(
+    resource_ids: Collection[int] | None = None,
+) -> Select[tuple[int]]:
     """Resources eligible to appear in a list: any non-repointed resource with an
     active membership. Membership-derived and ungated by licensing/source, so a
     resource whose licensed values are all absent still appears as a null-shell on
-    an unfiltered list (and drops out once a field is filtered)."""
+    an unfiltered list (and drops out once a field is filtered).
+
+    ``resource_ids`` narrows the universe to those resources."""
     m = MembershipModel
     r = ResourceModel
-    return (
+    stmt = (
         select(r.id.label("resource_id"))
         .select_from(r)
         .join(m, m.resource_id == r.id)
         .where(r.repointed_id.is_(None), m.status == MembershipStatus.ACTIVE)
         .distinct()
     )
+    if resource_ids is not None:
+        stmt = stmt.where(r.id.in_(list(dict.fromkeys(resource_ids))))
+    return stmt
 
 
 def base_resource_query(
     params: OGFieldQueryParams,
     licensed_sources: Collection[OGSISrcKey] | None = None,
+    resource_ids: Collection[int] | None = None,
 ) -> Select[tuple[int]]:
+    """Ids of the resources matching ``params``, in ``params``' sort order.
+
+    ``resource_ids`` limits the search to those resources, so callers that only
+    care about a known subset (e.g. the merge queue's resources) do not pivot
+    and rank every resource in the table.
+    """
     involved = _participating_columns(params)
-    universe = _resource_universe().cte("resource_universe")
+    universe = _resource_universe(resource_ids).cte("resource_universe")
 
     if not involved:
         # No value field filtered or sorted -> the universe alone (every active
         # resource); only the id filter below can narrow it.
         base = universe
         conditions: list[ColumnElement[bool]] = []
     else:
-        base_cte = construct_base_query_statement(licensed_sources)
+        base_cte = construct_base_query_statement(licensed_sources, resource_ids)
         ranked = add_ranking(base_cte).cte("ranked")
         pivot = _add_pivot_columns(
             select(ranked.c.resource_id.label("resource_id")),

deployments/api/src/stitch/api/entities.py

@@ -198,9 +198,16 @@ class MergeCandidateListItemView(MergeCandidateView):
 
 class MergeCandidateFilterParams(BaseModel):
     """Repeat ``status`` once per value (``?status=PENDING&status=DENIED``) to
-    match any of them. Omitted, every status is returned."""
+    match any of them. Omitted, every status is returned.
+
+    ``q`` matches a candidate when any of its resources -- or, once approved,
+    its merged resource -- matches the resource list's ``q`` search (substring
+    of the coalesced name, local name, basin, state/province or region). A
+    purely numeric ``q`` also matches those resources' ids.
+    """
 
     status: list[MergeCandidateStatus] | None = None
+    q: str | None = None
 
 
 class MergeCandidateSortParams(BaseModel):

deployments/api/tests/observability/test_query_name_actions.py

@@ -381,6 +381,20 @@ async def test_merge_candidate_action_labels(
             },
         )
 
+        captured_query_events.clear()
+        await mca.list_merge_candidates(session, MergeCandidateQueryParams(q="Cand"))
+        _assert_labels(
+            captured_query_events,
+            {
+                "merge_candidates.list.search_scope",
+                "merge_candidates.list.search",
+                "merge_candidates.list.count",
+                "merge_candidates.list",
+                "merge_candidates.list.status_counts",
+                "merge_candidates.list.names",
+            },
+        )
+
         captured_query_events.clear()
         await mca.get_merge_candidate(session, approved.id)
         _assert_labels(

deployments/api/tests/routers/test_merge_candidates_integration.py

@@ -15,6 +15,14 @@
     OGFieldResourceSourcePriority,
 )
 from stitch.ogsi.model import OGFieldResource, OGFieldSource
+from stitch.api.auth import get_token_claims
+from stitch.api.main import app
+from stitch.auth import TokenClaims
+from stitch.auth.permissions import (
+    MERGE_CANDIDATE_READ,
+    RESOURCE_READ,
+    SOURCE_READ_RMI,
+)
 
 
 async def _create_resource(
@@ -698,3 +706,187 @@ async def test_rejects_invalid_params(
             "/oil-gas-fields/merge-candidates", params=params
         )
         assert resp.status_code == 422, resp.text
+
+
+class TestMergeCandidateSearchIntegration:
+    """``q`` on the list: a candidate matches when any of its resources (or its
+    merged resource) matches the resource list's search, or a numeric ``q``
+    names one of those resources."""
+
+    @pytest.fixture
+    def make_resource(
+        self, integration_client, og_field_resource_factory, source_maker
+    ):
+        # The searchable fields default to blank, so only what a test sets can
+        # match (the factory would otherwise fill them with random text).
+        async def _make(*, source: str = "rmi", **fields) -> int:
+            searchable = {
+                "name_local": None,
+                "basin": None,
+                "state_province": None,
+                "region": None,
+            }
+            return await _create_resource_with_sources(
+                integration_client,
+                og_field_resource_factory,
+                [
+                    source_maker(
+                        source=source, managed=False, **{**searchable, **fields}
+                    )
+                ],
+            )
+
+        return _make
+
+    async def _search(self, client: AsyncClient, **params) -> list[int]:
+        resp = await client.get("/oil-gas-fields/merge-candidates", params=params)
+        assert resp.status_code == 200, resp.text
+        return [item["id"] for item in resp.json()["items"]]
+
+    @pytest.mark.anyio
+    async def test_matches_either_resource_by_name(
+        self, integration_client: AsyncClient, make_resource
+    ):
+        first = await make_resource(name="Zubair Main")
+        second = await make_resource(name="Zubayr Main")
+        candidate_id = await _create_candidate(integration_client, [first, second])
+        other = await _create_candidate(
+            integration_client,
+            [await make_resource(name="Ghawar"), await make_resource(name="Ghawar")],
+        )
+
+        # Case-insensitive substring, matching only the second resource.
+        assert await self._search(integration_client, q="zubayr") == [candidate_id]
+        assert await self._search(integration_client, q="Zub") == [candidate_id]
+        assert other not in await self._search(integration_client, q="zub")
+
+    @pytest.mark.anyio
+    async def test_matches_the_other_search_fields(
+        self, integration_client: AsyncClient, make_resource
+    ):
+        candidate_id = await _create_candidate(
+            integration_client,
+            [
+                await make_resource(name="Alpha", basin="Qwertian Basin"),
+                await make_resource(name="Alpha", region="Middle East"),
+            ],
+        )
+
+        assert await self._search(integration_client, q="qwertian") == [candidate_id]
+
+    @pytest.mark.anyio
+    async def test_finds_an_approved_candidate_by_its_merged_resource(
+        self, integration_client: AsyncClient, make_resource
+    ):
+        candidate_id = await _create_candidate(
+            integration_client,
+            [
+                await make_resource(name="Safaniya"),
+                await make_resource(name="Safaniya"),
+            ],
+        )
+        approve = await integration_client.post(
+            f"/oil-gas-fields/merge-candidates/{candidate_id}/approve"
+        )
+        assert approve.status_code == 200, approve.text
+
+        # The originals are emptied by the merge; only the merged resource
+        # still carries the name.
+        assert await self._search(integration_client, q="safaniya") == [candidate_id]
+
+    @pytest.mark.anyio
+    async def test_numeric_q_matches_resource_ids(
+        self, integration_client: AsyncClient, make_resource
+    ):
+        target = await make_resource(name="Unrelated")
+        by_id = await _create_candidate(
+            integration_client, [target, await make_resource(name="Unrelated")]
+        )
+        await _create_candidate(
+            integration_client,
+            [await make_resource(name="Ghawar"), await make_resource(name="Ghawar")],
+        )
+
+        assert await self._search(integration_client, q=str(target)) == [by_id]
+        # Candidate ids are not searched: no resource has this id.
+        assert await self._search(integration_client, q="987654321") == []
+        # Too long to be an id; still searched as text, without erroring.
+        assert await self._search(integration_client, q="9" * 30) == []
+
+    @pytest.mark.anyio
+    async def test_numeric_q_also_matches_digits_in_text(
+        self, integration_client: AsyncClient, make_resource
+    ):
+        by_text = await _create_candidate(
+            integration_client,
+            [
+                await make_resource(name="Block 990077"),
+                await make_resource(name="Block 990077"),
+            ],
+        )
+
+        # Far above any resource id here, so only the name can match.
+        assert await self._search(integration_client, q="990077") == [by_text]
+
+    @pytest.mark.anyio
+    async def test_combines_with_status_and_leaves_counts_whole_queue(
+        self, integration_client: AsyncClient, make_resource
+    ):
+        pending = await _create_candidate(
+            integration_client,
+            [await make_resource(name="Rumaila"), await make_resource(name="Rumaila")],
+        )
+        denied = await _create_candidate(
+            integration_client,
+            [await make_resource(name="Rumaila"), await make_resource(name="Rumaila")],
+        )
+        await _create_candidate(
+            integration_client,
+            [await make_resource(name="Ghawar"), await make_resource(name="Ghawar")],
+        )
+        resp = await integration_client.post(
+            f"/oil-gas-fields/merge-candidates/{denied}/deny"
+        )
+        assert resp.status_code == 200, resp.text
+
+        resp = await integration_client.get(
+            "/oil-gas-fields/merge-candidates",
+            params={"q": "rumaila", "status": "PENDING"},
+        )
+        body = resp.json()
+        assert [item["id"] for item in body["items"]] == [pending]
+        assert body["total_count"] == 1
+        assert body["status_counts"] == {"PENDING": 2, "APPROVED": 0, "DENIED": 1}
+
+    @pytest.mark.anyio
+    async def test_blank_q_is_no_search(
+        self, integration_client: AsyncClient, make_resource
+    ):
+        candidate_id = await _create_candidate(
+            integration_client,
+            [await make_resource(name="Ghawar"), await make_resource(name="Ghawar")],
+        )
+
+        assert await self._search(integration_client, q="   ") == [candidate_id]
+
+    @pytest.mark.anyio
+    async def test_ignores_names_from_unlicensed_sources(
+        self, integration_client: AsyncClient, make_resource
+    ):
+        await _create_candidate(
+            integration_client,
+            [
+                await make_resource(source="gem", name="Gemonly Field"),
+                await make_resource(source="gem", name="Gemonly Field"),
+            ],
+        )
+        app.dependency_overrides[get_token_claims] = lambda: TokenClaims(
+            sub="test|rmi-only",
+            email="rmi@test.com",
+            name="RMI Only",
+            permissions=frozenset(
+                {RESOURCE_READ, MERGE_CANDIDATE_READ, SOURCE_READ_RMI}
+            ),
+        )
+
+        assert await self._search(integration_client, q="gemonly") == []