Add governed Dataflow search source

src/govoplan_dataflow/backend/manifest.py

@@ -35,6 +35,7 @@ from govoplan_core.core.datasources import (
 from govoplan_core.core.policy import (
     CAPABILITY_POLICY_DEFINITION_GOVERNANCE,
 )
+from govoplan_core.core.search import SearchSourceProviderRegistration
 from govoplan_core.core.views import ViewSurface
 from govoplan_core.db.base import Base
 from govoplan_dataflow.backend.db import models as dataflow_models
@@ -196,6 +197,14 @@ def _trigger_provider(context: ModuleContext):
     return SqlDataflowTriggerDispatcher(registry=context.registry)
 
 
+def _search_source_provider(context: ModuleContext):
+    from govoplan_dataflow.backend.search_source import (
+        create_dataflow_search_source,
+    )
+
+    return create_dataflow_search_source(context)
+
+
 def _tenant_summary(session, tenant_id: str) -> dict[str, int]:
     return {
         "dataflow_pipelines": (
@@ -241,6 +250,7 @@ manifest = ModuleManifest(
         "policy",
         "reporting",
         "risk_compliance",
+        "search",
         "workflow",
     ),
     optional_capabilities=(
@@ -292,6 +302,12 @@ manifest = ModuleManifest(
             version_max_exclusive="1.0.0",
             optional=True,
         ),
+        ModuleInterfaceRequirement(
+            name="search.source",
+            version_min="1.0.0",
+            version_max_exclusive="2.0.0",
+            optional=True,
+        ),
     ),
     permissions=PERMISSIONS,
     role_templates=ROLE_TEMPLATES,
@@ -339,6 +355,12 @@ manifest = ModuleManifest(
         CAPABILITY_DATAFLOW_RUN_LIFECYCLE: _run_provider,
         CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER: _trigger_provider,
     },
+    search_sources=(
+        SearchSourceProviderRegistration(
+            id="dataflow.pipelines",
+            factory=_search_source_provider,
+        ),
+    ),
     tenant_summary_providers=(_tenant_summary,),
     migration_spec=MigrationSpec(
         module_id=MODULE_ID,

src/govoplan_dataflow/backend/search_source.py

@@ -0,0 +1,226 @@
+from __future__ import annotations
+
+from collections.abc import Mapping, Sequence
+from urllib.parse import quote
+
+from sqlalchemy import func, or_, select
+from sqlalchemy.orm import Session
+
+from govoplan_core.auth import ApiPrincipal
+from govoplan_core.core.modules import ModuleContext
+from govoplan_core.core.search import (
+    SearchAuthorizationRequest,
+    SearchBackfillPage,
+    SearchBackfillRequest,
+    SearchDocument,
+    SearchResourceType,
+)
+from govoplan_dataflow.backend.db.models import DataflowPipeline
+from govoplan_dataflow.backend.governance import definition_decision
+
+
+PROVIDER_ID = "dataflow.pipelines"
+RESOURCE_TYPE = "dataflow_pipeline"
+READ_SCOPE = "dataflow:pipeline:read"
+ADMIN_SCOPE = "dataflow:pipeline:admin"
+
+
+class DataflowSearchSource:
+    def __init__(self, registry: object | None) -> None:
+        self.registry = registry
+
+    def resource_types(self) -> Sequence[SearchResourceType]:
+        return (
+            SearchResourceType(
+                provider_id=PROVIDER_ID,
+                module_id="dataflow",
+                resource_type=RESOURCE_TYPE,
+                label="Data pipelines",
+                requires_authorization_recheck=True,
+            ),
+        )
+
+    def backfill(
+        self,
+        session: object,
+        *,
+        request: SearchBackfillRequest,
+    ) -> SearchBackfillPage:
+        db = _session(session)
+        if (
+            request.provider_id != PROVIDER_ID
+            or request.resource_type != RESOURCE_TYPE
+        ):
+            raise ValueError("Unsupported Dataflow search source.")
+
+        source_filter = or_(
+            DataflowPipeline.tenant_id == request.tenant_id,
+            DataflowPipeline.tenant_id.is_(None),
+        )
+        statement = select(DataflowPipeline).where(
+            source_filter,
+            DataflowPipeline.deleted_at.is_(None),
+        )
+        if request.cursor:
+            statement = statement.where(
+                DataflowPipeline.id > request.cursor
+            )
+        rows = list(
+            db.scalars(
+                statement.order_by(DataflowPipeline.id).limit(
+                    request.limit + 1
+                )
+            )
+        )
+        has_more = len(rows) > request.limit
+        selected = rows[: request.limit]
+        high_watermark = db.scalar(
+            select(func.max(DataflowPipeline.updated_at)).where(
+                source_filter,
+                DataflowPipeline.deleted_at.is_(None),
+            )
+        )
+        return SearchBackfillPage(
+            documents=tuple(
+                _pipeline_document(
+                    pipeline,
+                    tenant_id=request.tenant_id,
+                )
+                for pipeline in selected
+            ),
+            next_cursor=selected[-1].id if has_more and selected else None,
+            complete=not has_more,
+            high_watermark=(
+                high_watermark.isoformat()
+                if high_watermark is not None
+                else None
+            ),
+        )
+
+    def authorize(
+        self,
+        session: object,
+        principal: object,
+        *,
+        requests: Sequence[SearchAuthorizationRequest],
+    ) -> Mapping[str, bool]:
+        decisions = {
+            request.reference.key: False
+            for request in requests
+        }
+        if not isinstance(principal, ApiPrincipal):
+            return decisions
+        if not (
+            principal.has(READ_SCOPE)
+            or principal.has(ADMIN_SCOPE)
+        ):
+            return decisions
+
+        valid_requests = tuple(
+            request
+            for request in requests
+            if (
+                request.reference.tenant_id == principal.tenant_id
+                and request.reference.module_id == "dataflow"
+                and request.reference.resource_type == RESOURCE_TYPE
+            )
+        )
+        if not valid_requests:
+            return decisions
+
+        db = _session(session)
+        ids = {
+            request.reference.resource_id
+            for request in valid_requests
+        }
+        pipelines = {
+            pipeline.id: pipeline
+            for pipeline in db.scalars(
+                select(DataflowPipeline).where(
+                    DataflowPipeline.id.in_(ids),
+                    or_(
+                        DataflowPipeline.tenant_id
+                        == principal.tenant_id,
+                        DataflowPipeline.tenant_id.is_(None),
+                    ),
+                    DataflowPipeline.deleted_at.is_(None),
+                )
+            )
+        }
+        for request in valid_requests:
+            pipeline = pipelines.get(
+                request.reference.resource_id
+            )
+            if pipeline is None:
+                continue
+            try:
+                allowed = definition_decision(
+                    pipeline,
+                    principal=principal,
+                    registry=self.registry,
+                    action="view",
+                ).allowed
+            except (PermissionError, RuntimeError, ValueError):
+                allowed = False
+            decisions[request.reference.key] = allowed
+        return decisions
+
+
+def create_dataflow_search_source(
+    context: ModuleContext,
+) -> DataflowSearchSource:
+    return DataflowSearchSource(context.registry)
+
+
+def _pipeline_document(
+    pipeline: DataflowPipeline,
+    *,
+    tenant_id: str,
+) -> SearchDocument:
+    updated_at = pipeline.updated_at
+    source_revision = (
+        f"{pipeline.current_revision}:"
+        f"{updated_at.isoformat() if updated_at is not None else 'unknown'}"
+    )
+    return SearchDocument(
+        tenant_id=tenant_id,
+        module_id="dataflow",
+        provider_id=PROVIDER_ID,
+        resource_type=RESOURCE_TYPE,
+        resource_id=pipeline.id,
+        title=pipeline.name,
+        url=f"/dataflow?pipelineId={quote(pipeline.id, safe='')}",
+        summary=pipeline.description,
+        body=pipeline.description,
+        keywords=(
+            pipeline.status,
+            pipeline.scope_type,
+            pipeline.definition_kind,
+        ),
+        visibility="tenant",
+        source_revision=source_revision,
+        source_updated_at=updated_at,
+        metadata={
+            "status": pipeline.status,
+            "scope_type": pipeline.scope_type,
+            "definition_kind": pipeline.definition_kind,
+            "current_revision": pipeline.current_revision,
+        },
+        requires_authorization_recheck=True,
+    )
+
+
+def _session(value: object) -> Session:
+    if not isinstance(value, Session):
+        raise TypeError(
+            "Dataflow search requires a SQLAlchemy session."
+        )
+    return value
+
+
+__all__ = [
+    "DataflowSearchSource",
+    "PROVIDER_ID",
+    "RESOURCE_TYPE",
+    "create_dataflow_search_source",
+]

tests/test_search_source.py

@@ -0,0 +1,179 @@
+from __future__ import annotations
+
+from types import SimpleNamespace
+import unittest
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+from govoplan_core.auth import ApiPrincipal
+from govoplan_core.core.access import PrincipalRef
+from govoplan_core.core.search import (
+    SearchAuthorizationRequest,
+    SearchBackfillRequest,
+    SearchResourceReference,
+)
+from govoplan_core.db.base import Base
+from govoplan_dataflow.backend.db.models import DataflowPipeline
+from govoplan_dataflow.backend.manifest import (
+    ADMIN_SCOPE,
+    READ_SCOPE,
+    get_manifest,
+)
+from govoplan_dataflow.backend.search_source import (
+    DataflowSearchSource,
+    PROVIDER_ID,
+    RESOURCE_TYPE,
+)
+
+
+class DataflowSearchSourceTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite://")
+        Base.metadata.create_all(
+            self.engine,
+            tables=(DataflowPipeline.__table__,),
+        )
+        self.session = Session(self.engine)
+        self.session.add_all(
+            (
+                _pipeline(
+                    "00000000-0000-0000-0000-000000000001",
+                    "tenant-1",
+                    "Monthly reconciliation",
+                ),
+                _pipeline(
+                    "00000000-0000-0000-0000-000000000002",
+                    "tenant-1",
+                    "Sanctions screening",
+                ),
+                _pipeline(
+                    "00000000-0000-0000-0000-000000000003",
+                    "tenant-2",
+                    "Other tenant",
+                ),
+            )
+        )
+        self.session.commit()
+        self.source = DataflowSearchSource(registry=None)
+
+    def tearDown(self) -> None:
+        self.session.close()
+        self.engine.dispose()
+
+    def test_manifest_announces_optional_search_source(self) -> None:
+        manifest = get_manifest()
+
+        self.assertIn("search", manifest.optional_dependencies)
+        self.assertEqual(
+            (PROVIDER_ID,),
+            tuple(item.id for item in manifest.search_sources),
+        )
+
+    def test_backfill_is_tenant_bounded_and_stably_paged(self) -> None:
+        first = self.source.backfill(
+            self.session,
+            request=SearchBackfillRequest(
+                tenant_id="tenant-1",
+                provider_id=PROVIDER_ID,
+                resource_type=RESOURCE_TYPE,
+                rebuild_id="rebuild-1",
+                limit=1,
+            ),
+        )
+        second = self.source.backfill(
+            self.session,
+            request=SearchBackfillRequest(
+                tenant_id="tenant-1",
+                provider_id=PROVIDER_ID,
+                resource_type=RESOURCE_TYPE,
+                rebuild_id="rebuild-1",
+                cursor=first.next_cursor,
+                limit=1,
+            ),
+        )
+
+        self.assertFalse(first.complete)
+        self.assertTrue(second.complete)
+        self.assertEqual(
+            {"Monthly reconciliation", "Sanctions screening"},
+            {
+                first.documents[0].title,
+                second.documents[0].title,
+            },
+        )
+        self.assertTrue(
+            first.documents[0].url.startswith(
+                "/dataflow?pipelineId="
+            )
+        )
+        self.assertTrue(
+            first.documents[0].requires_authorization_recheck
+        )
+
+    def test_authorization_rechecks_tenant_scope_and_permission(self) -> None:
+        reference = SearchResourceReference(
+            tenant_id="tenant-1",
+            module_id="dataflow",
+            resource_type=RESOURCE_TYPE,
+            resource_id="00000000-0000-0000-0000-000000000001",
+        )
+        request = SearchAuthorizationRequest(
+            reference=reference,
+            source_revision="1",
+        )
+
+        allowed = self.source.authorize(
+            self.session,
+            _principal({READ_SCOPE}),
+            requests=(request,),
+        )
+        denied = self.source.authorize(
+            self.session,
+            _principal(set()),
+            requests=(request,),
+        )
+        administrative = self.source.authorize(
+            self.session,
+            _principal({ADMIN_SCOPE}),
+            requests=(request,),
+        )
+
+        self.assertTrue(allowed[reference.key])
+        self.assertFalse(denied[reference.key])
+        self.assertTrue(administrative[reference.key])
+
+
+def _pipeline(
+    pipeline_id: str,
+    tenant_id: str,
+    name: str,
+) -> DataflowPipeline:
+    return DataflowPipeline(
+        id=pipeline_id,
+        tenant_id=tenant_id,
+        scope_type="tenant",
+        scope_id=tenant_id,
+        definition_kind="flow",
+        name=name,
+        description=f"{name} description",
+        status="active",
+        current_revision=1,
+    )
+
+
+def _principal(scopes: set[str]) -> ApiPrincipal:
+    return ApiPrincipal(
+        principal=PrincipalRef(
+            account_id="account-1",
+            membership_id="membership-1",
+            tenant_id="tenant-1",
+            scopes=frozenset(scopes),
+        ),
+        account=SimpleNamespace(id="account-1"),
+        user=SimpleNamespace(id="membership-1"),
+    )
+
+
+if __name__ == "__main__":
+    unittest.main()

webui/src/features/dataflow/DataflowPage.tsx

@@ -43,6 +43,7 @@ import {
   type AuthInfo
 } from "@govoplan/core-webui";
 import { ReactFlowProvider } from "@xyflow/react";
+import { useLocation } from "react-router-dom";
 import {
   compileDataflowSql,
   createDataflowTrigger,
@@ -99,6 +100,11 @@ const AUTO_PREVIEW_STORAGE_KEY = "govoplan.dataflow.auto-preview";
 
 export default function DataflowPage({ settings, auth }: { settings: ApiSettings; auth: AuthInfo }) {
   const { requestNavigation, requestDiscard } = useUnsavedChanges();
+  const location = useLocation();
+  const requestedPipelineId = useMemo(
+    () => new URLSearchParams(location.search).get("pipelineId"),
+    [location.search]
+  );
   const [pipelines, setPipelines] = useState<Pipeline[]>([]);
   const [draft, setDraft] = useState<PipelineDraft | null>(null);
   const [savedDraft, setSavedDraft] = useState<PipelineDraft | null>(null);
@@ -219,8 +225,8 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings
   }, [draft?.id, settings]);
 
   useEffect(() => {
-    void loadPipelines();
+    void loadPipelines(requestedPipelineId);
-  }, []);
+  }, [requestedPipelineId]);
 
   useEffect(() => {
     let cancelled = false;