Add permission-aware datasource search source

src/govoplan_datasources/backend/manifest.py

@@ -35,9 +35,13 @@ from govoplan_core.core.provider_governance import (
     ModuleArchitectureDocumentation,
     ModuleMaturityEvidence,
 )
+from govoplan_core.core.search import SearchSourceProviderRegistration
 from govoplan_core.core.views import ViewSurface
 from govoplan_core.db.base import Base
 from govoplan_datasources.backend.db import models as datasource_models
+from govoplan_datasources.backend.search_source import (
+    create_datasources_search_source,
+)
 from govoplan_datasources.backend.service import (
     ADMIN_SCOPE,
     CATALOGUE_READ_SCOPE,
@@ -219,6 +223,7 @@ manifest = ModuleManifest(
         "files",
         "notifications",
         "policy",
+        "search",
     ),
     optional_capabilities=(
         CAPABILITY_AUTH_PRINCIPAL_RESOLVER,
@@ -254,6 +259,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,
@@ -302,6 +313,12 @@ manifest = ModuleManifest(
         CAPABILITY_DATASOURCE_PUBLICATION: _provider,
     },
     tenant_summary_providers=(_tenant_summary,),
+    search_sources=(
+        SearchSourceProviderRegistration(
+            id="datasources.catalogue",
+            factory=create_datasources_search_source,
+        ),
+    ),
     migration_spec=MigrationSpec(
         module_id=MODULE_ID,
         metadata=Base.metadata,
@@ -350,7 +367,11 @@ manifest = ModuleManifest(
                 "reports, controls, and decisions. It preserves origin source mode, "
                 "structured health, declared pushdown, and effective row, byte, and time "
                 "limits for live previews. Governance metadata visibility does not "
-                "grant access to protected rows."
+                "grant access to protected rows. When Search is enabled, the module "
+                "indexes only bounded catalogue labels and governance-safe facets, "
+                "then rechecks the current catalogue permission before returning a "
+                "result. Rows, schemas, connector references, credentials, arbitrary "
+                "metadata, and provenance are never copied into the search index."
             ),
             layer="available",
             documentation_types=("admin", "user"),
@@ -362,6 +383,7 @@ manifest = ModuleManifest(
                 "files",
                 "reporting",
                 "risk_compliance",
+                "search",
             ),
             order=70,
             metadata={

src/govoplan_datasources/backend/search_source.py

@@ -0,0 +1,179 @@
+from __future__ import annotations
+
+from collections.abc import Mapping, Sequence
+from urllib.parse import quote
+
+from sqlalchemy import func, 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_datasources.backend.db.models import DatasourceRecord
+from govoplan_datasources.backend.service import ADMIN_SCOPE, CATALOGUE_READ_SCOPE
+
+
+PROVIDER_ID = "datasources.catalogue"
+RESOURCE_TYPE = "datasource"
+
+
+class DatasourcesSearchSource:
+    def resource_types(self) -> Sequence[SearchResourceType]:
+        return (
+            SearchResourceType(
+                provider_id=PROVIDER_ID,
+                module_id="datasources",
+                resource_type=RESOURCE_TYPE,
+                label="Datasources",
+                requires_authorization_recheck=True,
+            ),
+        )
+
+    def backfill(
+        self,
+        session: object,
+        *,
+        request: SearchBackfillRequest,
+    ) -> SearchBackfillPage:
+        _assert_source(request.provider_id, request.resource_type)
+        db = _session(session)
+        statement = select(DatasourceRecord).where(
+            DatasourceRecord.tenant_id == request.tenant_id,
+            DatasourceRecord.deleted_at.is_(None),
+        )
+        if request.cursor:
+            statement = statement.where(DatasourceRecord.id > request.cursor)
+        rows = list(
+            db.scalars(
+                statement.order_by(DatasourceRecord.id).limit(request.limit + 1)
+            ).all()
+        )
+        has_more = len(rows) > request.limit
+        selected = rows[: request.limit]
+        high_watermark = db.scalar(
+            select(func.max(DatasourceRecord.updated_at)).where(
+                DatasourceRecord.tenant_id == request.tenant_id,
+                DatasourceRecord.deleted_at.is_(None),
+            )
+        )
+        return SearchBackfillPage(
+            documents=tuple(_document(row) for row 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 else None,
+        )
+
+    def authorize(
+        self,
+        session: object,
+        principal: object,
+        *,
+        requests: Sequence[SearchAuthorizationRequest],
+    ) -> Mapping[str, bool]:
+        decisions = {item.reference.key: False for item in requests}
+        if not isinstance(principal, ApiPrincipal) or not (
+            principal.has(CATALOGUE_READ_SCOPE) or principal.has(ADMIN_SCOPE)
+        ):
+            return decisions
+        db = _session(session)
+        eligible = [
+            request
+            for request in requests
+            if request.reference.tenant_id == principal.tenant_id
+            and request.reference.module_id == "datasources"
+            and request.reference.resource_type == RESOURCE_TYPE
+        ]
+        resource_ids = {request.reference.resource_id for request in eligible}
+        available_ids = (
+            set(
+                db.scalars(
+                    select(DatasourceRecord.id).where(
+                        DatasourceRecord.tenant_id == principal.tenant_id,
+                        DatasourceRecord.id.in_(resource_ids),
+                        DatasourceRecord.deleted_at.is_(None),
+                    )
+                ).all()
+            )
+            if resource_ids
+            else set()
+        )
+        for request in eligible:
+            decisions[request.reference.key] = (
+                request.reference.resource_id in available_ids
+            )
+        return decisions
+
+
+def create_datasources_search_source(
+    _context: ModuleContext,
+) -> DatasourcesSearchSource:
+    return DatasourcesSearchSource()
+
+
+def _document(row: DatasourceRecord) -> SearchDocument:
+    datasource_ref = quote(f"datasource:{row.id}", safe="")
+    return SearchDocument(
+        tenant_id=row.tenant_id,
+        module_id="datasources",
+        provider_id=PROVIDER_ID,
+        resource_type=RESOURCE_TYPE,
+        resource_id=row.id,
+        title=row.name,
+        url=f"/datasources?datasource={datasource_ref}",
+        summary=(row.description or row.source_name)[:4000],
+        keywords=tuple(
+            value[:200]
+            for value in (
+                row.source_name,
+                row.kind,
+                row.mode,
+                row.shape,
+                row.status,
+                row.classification,
+                row.publication_state,
+            )
+            if value
+        ),
+        visibility="restricted",
+        acl_tokens=(
+            f"scope:{CATALOGUE_READ_SCOPE}",
+            f"scope:{ADMIN_SCOPE}",
+        ),
+        metadata={
+            "kind": row.kind,
+            "mode": row.mode,
+            "shape": row.shape,
+            "status": row.status,
+            "classification": row.classification,
+            "publication_state": row.publication_state,
+            "authority_mode": row.authority_mode,
+        },
+        source_revision=f"{row.schema_version}:{row.updated_at.isoformat()}",
+        source_updated_at=row.updated_at,
+        requires_authorization_recheck=True,
+    )
+
+
+def _assert_source(provider_id: str, resource_type: str) -> None:
+    if provider_id != PROVIDER_ID or resource_type != RESOURCE_TYPE:
+        raise ValueError("Unsupported Datasources search source.")
+
+
+def _session(value: object) -> Session:
+    if not isinstance(value, Session):
+        raise TypeError("Datasources search requires a SQLAlchemy session.")
+    return value
+
+
+__all__ = [
+    "DatasourcesSearchSource",
+    "PROVIDER_ID",
+    "RESOURCE_TYPE",
+    "create_datasources_search_source",
+]