Index authorized mailbox cache changes in Search

src/govoplan_mail/backend/mailbox_index.py

@@ -7,11 +7,17 @@ from threading import Lock
 from sqlalchemy import func
 from sqlalchemy.orm import Session
 
+from govoplan_core.core.change_sequence import record_change
 from govoplan_core.security.time import ensure_aware_utc, utc_now
-from govoplan_mail.backend.db.models import MailMailboxFolderIndex, MailMailboxMessageIndex
+from govoplan_mail.backend.db.models import (
+    MailMailboxFolderIndex,
+    MailMailboxMessageIndex,
+    new_uuid,
+)
 from govoplan_mail.backend.sending.imap import ImapFolderListResult, ImapMailboxInfo, ImapMailboxMessageListResult, ImapMailboxMessageSummary
 
 MAILBOX_INDEX_TTL_SECONDS = 30
+MAILBOX_MESSAGES_COLLECTION = "mail.mailbox_messages"
 
 _refresh_lock = Lock()
 _refreshing_keys: set[tuple[str, str, str]] = set()
@@ -44,6 +50,13 @@ def clear_mailbox_index(session: Session, *, profile_id: str) -> tuple[int, int]
     transport change. Message rows are removed before their folder metadata.
     """
 
+    messages = (
+        session.query(MailMailboxMessageIndex)
+        .filter(MailMailboxMessageIndex.profile_id == profile_id)
+        .all()
+    )
+    for message in messages:
+        _record_message_change(session, message, operation="deleted")
     deleted_messages = (
         session.query(MailMailboxMessageIndex)
         .filter(MailMailboxMessageIndex.profile_id == profile_id)
@@ -89,6 +102,17 @@ def cache_mailbox_folders(
     expected_names = {folder.name for folder in result.folders}
     removed_names = set(existing) - expected_names
     if removed_names:
+        removed_messages = (
+            session.query(MailMailboxMessageIndex)
+            .filter(
+                MailMailboxMessageIndex.tenant_id == tenant_id,
+                MailMailboxMessageIndex.profile_id == profile_id,
+                MailMailboxMessageIndex.folder.in_(removed_names),
+            )
+            .all()
+        )
+        for message in removed_messages:
+            _record_message_change(session, message, operation="deleted")
         (
             session.query(MailMailboxMessageIndex)
             .filter(
@@ -139,6 +163,17 @@ def cache_mailbox_messages(
 
     uids = [message.uid for message in result.messages]
     if result.total_count <= 0:
+        removed_messages = (
+            session.query(MailMailboxMessageIndex)
+            .filter(
+                MailMailboxMessageIndex.tenant_id == tenant_id,
+                MailMailboxMessageIndex.profile_id == profile_id,
+                MailMailboxMessageIndex.folder == result.folder,
+            )
+            .all()
+        )
+        for message in removed_messages:
+            _record_message_change(session, message, operation="deleted")
         (
             session.query(MailMailboxMessageIndex)
             .filter(
@@ -161,6 +196,7 @@ def cache_mailbox_messages(
         if uids:
             stale_query = stale_query.filter(MailMailboxMessageIndex.uid.notin_(uids))
         for row in stale_query.all():
+            _record_message_change(session, row, operation="deleted")
             session.delete(row)
     existing = {}
     if uids:
@@ -177,13 +213,16 @@ def cache_mailbox_messages(
         }
     for index, message in enumerate(result.messages):
         row = existing.get(message.uid)
+        created = row is None
         if row is None:
             row = MailMailboxMessageIndex(
+                id=new_uuid(),
                 tenant_id=tenant_id,
                 profile_id=profile_id,
                 folder=result.folder,
                 uid=message.uid,
             )
+        previous = None if created else _message_state(row)
         row.uid_int = _uid_int(message.uid)
         row.sort_position = result.offset + index
         row.subject = message.subject
@@ -198,6 +237,12 @@ def cache_mailbox_messages(
         row.attachment_count = message.attachment_count
         row.indexed_at = indexed_at
         session.add(row)
+        if created or previous != _message_state(row):
+            _record_message_change(
+                session,
+                row,
+                operation="created" if created else "updated",
+            )
 
 
 def cached_mailbox_folders(
@@ -308,6 +353,48 @@ def _message_from_index(row: MailMailboxMessageIndex) -> ImapMailboxMessageSumma
     )
 
 
+def _message_state(row: MailMailboxMessageIndex) -> tuple[object, ...]:
+    return (
+        row.folder,
+        row.uid,
+        row.sort_position,
+        row.subject,
+        row.from_header,
+        row.to_header,
+        row.cc_header,
+        row.date,
+        row.message_id,
+        tuple(row.flags or ()),
+        row.size_bytes,
+        row.body_preview,
+        row.attachment_count,
+    )
+
+
+def _record_message_change(
+    session: Session,
+    row: MailMailboxMessageIndex,
+    *,
+    operation: str,
+) -> None:
+    record_change(
+        session,
+        module_id="mail",
+        collection=MAILBOX_MESSAGES_COLLECTION,
+        resource_type="mailbox_message",
+        resource_id=row.id,
+        operation=operation,
+        tenant_id=row.tenant_id,
+        actor_type="system",
+        payload={
+            "profile_id": row.profile_id,
+            "folder": row.folder,
+            "uid": row.uid,
+            "message_id": row.message_id,
+        },
+    )
+
+
 def _uid_int(uid: str) -> int:
     try:
         return int(str(uid))

src/govoplan_mail/backend/manifest.py

@@ -35,6 +35,7 @@ from govoplan_core.core.provider_governance import (
     ProviderObjectDeclaration,
     declared_module_architecture,
 )
+from govoplan_core.core.search import SearchSourceProviderRegistration
 from govoplan_core.core.views import ViewSurface
 from govoplan_core.db.base import Base
 from govoplan_mail.backend.documentation import (
@@ -48,6 +49,7 @@ from govoplan_mail.backend.provider_state import (
     smtp_provider_states,
 )
 from govoplan_mail.backend.db import models as mail_models  # noqa: F401 - populate Mail ORM metadata
+from govoplan_mail.backend.search_source import create_mail_search_source
 
 
 _mail_table_retirement_provider = drop_table_retirement_provider(
@@ -299,7 +301,7 @@ manifest = ModuleManifest(
     name="Mail",
     version="0.1.10",
     required_capabilities=(CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR),
-    optional_dependencies=("campaigns", "addresses", "calendar"),
+    optional_dependencies=("campaigns", "addresses", "calendar", "search"),
     provides_interfaces=(
         ModuleInterfaceProvider(name="mail.campaign_delivery", version="0.2.0"),
         ModuleInterfaceProvider(name="mail.delivery_commands", version="0.1.0"),
@@ -332,10 +334,22 @@ manifest = ModuleManifest(
             version_max_exclusive="0.2.0",
             optional=True,
         ),
+        ModuleInterfaceRequirement(
+            name="search.source",
+            version_min="1.0.0",
+            version_max_exclusive="2.0.0",
+            optional=True,
+        ),
     ),
     permissions=PERMISSIONS,
     route_factory=_mail_router,
     role_templates=ROLE_TEMPLATES,
+    search_sources=(
+        SearchSourceProviderRegistration(
+            id="mail.mailbox_messages",
+            factory=create_mail_search_source,
+        ),
+    ),
     nav_items=(NavItem(path="/mail", label="Mail", icon="mail", required_any=("mail:mailbox:read", "mail:bounce:read", "mail:bounce:manage"), order=50),),
     frontend=FrontendModule(
         module_id="mail",
@@ -404,6 +418,23 @@ manifest = ModuleManifest(
         ).SqlMailBounceProcessingProvider(),
     },
     documentation=(
+        DocumentationTopic(
+            id="mail.search.mailbox-messages",
+            title="Search authorized mailbox messages",
+            summary="Expose bounded cached mailbox message metadata and previews to permission-aware platform Search.",
+            body=(
+                "When Search is installed, Mail contributes its read-only mailbox cache, never credentials or "
+                "unbounded raw messages. Results require both mailbox-read and profile-use authority and recheck "
+                "the current profile scope, policy, activity, tenant, user, group, or Campaign visibility before "
+                "returning a result. Mailbox refreshes publish durable index changes; a rebuild remains available "
+                "for reconciliation."
+            ),
+            layer="configured",
+            documentation_types=("admin", "user"),
+            audience=("mail_user", "mail_admin", "administrator"),
+            related_modules=("search",),
+            order=38,
+        ),
         DocumentationTopic(
             id="mail.profiles-and-policy",
             title="Mail profiles and policy hierarchy",

src/govoplan_mail/backend/search_source.py

@@ -0,0 +1,292 @@
+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.events import PlatformEvent
+from govoplan_core.core.modules import ModuleContext
+from govoplan_core.core.search import (
+    SearchAuthorizationRequest,
+    SearchBackfillPage,
+    SearchBackfillRequest,
+    SearchDocument,
+    SearchIndexChange,
+    SearchResourceReference,
+    SearchResourceType,
+)
+from govoplan_mail.backend.db.models import (
+    MailMailboxMessageIndex,
+    MailServerProfile,
+)
+from govoplan_mail.backend.mail_profiles import mail_profile_visible_to_actor
+
+
+PROVIDER_ID = "mail.mailbox_messages"
+RESOURCE_TYPE = "mailbox_message"
+READ_SCOPE = "mail:mailbox:read"
+USE_SCOPE = "mail:profile:use"
+
+
+class MailSearchSource:
+    def resource_types(self) -> Sequence[SearchResourceType]:
+        return (
+            SearchResourceType(
+                provider_id=PROVIDER_ID,
+                module_id="mail",
+                resource_type=RESOURCE_TYPE,
+                label="Mailbox messages",
+                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(MailMailboxMessageIndex, MailServerProfile)
+            .join(
+                MailServerProfile,
+                MailServerProfile.id == MailMailboxMessageIndex.profile_id,
+            )
+            .where(MailMailboxMessageIndex.tenant_id == request.tenant_id)
+        )
+        if request.cursor:
+            statement = statement.where(
+                MailMailboxMessageIndex.id > request.cursor
+            )
+        rows = list(
+            db.execute(
+                statement.order_by(MailMailboxMessageIndex.id).limit(
+                    request.limit + 1
+                )
+            ).all()
+        )
+        has_more = len(rows) > request.limit
+        selected = rows[: request.limit]
+        high_watermark = db.scalar(
+            select(func.max(MailMailboxMessageIndex.indexed_at)).where(
+                MailMailboxMessageIndex.tenant_id == request.tenant_id
+            )
+        )
+        return SearchBackfillPage(
+            documents=tuple(
+                _document(message, profile=profile)
+                for message, profile in selected
+            ),
+            next_cursor=(
+                selected[-1][0].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 = {item.reference.key: False for item in requests}
+        if not isinstance(principal, ApiPrincipal) or not (
+            principal.has(READ_SCOPE) and principal.has(USE_SCOPE)
+        ):
+            return decisions
+        valid = tuple(
+            item
+            for item in requests
+            if item.reference.tenant_id == principal.tenant_id
+            and item.reference.module_id == "mail"
+            and item.reference.resource_type == RESOURCE_TYPE
+        )
+        if not valid:
+            return decisions
+        db = _session(session)
+        ids = {item.reference.resource_id for item in valid}
+        messages = {
+            message.id: (message, profile)
+            for message, profile in db.execute(
+                select(MailMailboxMessageIndex, MailServerProfile)
+                .join(
+                    MailServerProfile,
+                    MailServerProfile.id
+                    == MailMailboxMessageIndex.profile_id,
+                )
+                .where(
+                    MailMailboxMessageIndex.id.in_(ids),
+                    MailMailboxMessageIndex.tenant_id == principal.tenant_id,
+                )
+            ).all()
+        }
+        user_id = str(
+            getattr(principal.user, "id", "") or principal.membership_id or ""
+        )
+        for item in valid:
+            match = messages.get(item.reference.resource_id)
+            if match is None:
+                continue
+            _message, profile = match
+            try:
+                allowed = mail_profile_visible_to_actor(
+                    db,
+                    profile=profile,
+                    tenant_id=principal.tenant_id,
+                    user_id=user_id,
+                    group_ids=principal.group_ids,
+                    tenant_admin=principal.has("tenant:*"),
+                    require_active=True,
+                )
+            except (RuntimeError, ValueError):
+                allowed = False
+            decisions[item.reference.key] = allowed
+        return decisions
+
+    def index_changes_for_event(
+        self,
+        session: object,
+        *,
+        event: PlatformEvent,
+        delivery_key: str,
+    ) -> Sequence[SearchIndexChange]:
+        if (
+            event.module_id != "mail"
+            or event.tenant is None
+            or event.resource is None
+            or event.resource.type != RESOURCE_TYPE
+            or event.resource.id is None
+        ):
+            return ()
+        db = _session(session)
+        row = db.get(MailMailboxMessageIndex, event.resource.id)
+        profile = (
+            db.get(MailServerProfile, row.profile_id)
+            if row is not None and row.tenant_id == event.tenant.id
+            else None
+        )
+        deleted = (
+            row is None
+            or row.tenant_id != event.tenant.id
+            or profile is None
+        )
+        cursor = event.event_id
+        document = (
+            None
+            if deleted
+            else _document(row, profile=profile, change_cursor=cursor)
+        )
+        reference = SearchResourceReference(
+            tenant_id=event.tenant.id,
+            module_id="mail",
+            resource_type=RESOURCE_TYPE,
+            resource_id=event.resource.id,
+        )
+        return (
+            SearchIndexChange(
+                change_id=f"{delivery_key}:{PROVIDER_ID}",
+                provider_id=PROVIDER_ID,
+                kind="delete" if deleted else "upsert",
+                reference=reference,
+                source_revision=(
+                    document.source_revision if document is not None else cursor
+                ),
+                cursor=cursor,
+                document=document,
+                occurred_at=event.occurred_at,
+            ),
+        )
+
+
+def create_mail_search_source(_context: ModuleContext) -> MailSearchSource:
+    return MailSearchSource()
+
+
+def _document(
+    message: MailMailboxMessageIndex,
+    *,
+    profile: MailServerProfile,
+    change_cursor: str | None = None,
+) -> SearchDocument:
+    tokens = [f"scope:{READ_SCOPE}", f"scope:{USE_SCOPE}"]
+    scope_type = str(profile.scope_type or "tenant")
+    if scope_type == "user" and profile.scope_id:
+        tokens.append(f"membership:{profile.scope_id}")
+    elif scope_type == "group" and profile.scope_id:
+        tokens.append(f"group:{profile.scope_id}")
+    title = (message.subject or "(No subject)")[:500]
+    summary = " | ".join(
+        value for value in (message.from_header, message.date) if value
+    )[:4000]
+    body = " ".join(
+        value
+        for value in (
+            message.from_header,
+            message.to_header,
+            message.cc_header,
+            message.body_preview,
+        )
+        if value
+    )[:200_000]
+    updated_at = message.updated_at or message.indexed_at
+    return SearchDocument(
+        tenant_id=message.tenant_id,
+        module_id="mail",
+        provider_id=PROVIDER_ID,
+        resource_type=RESOURCE_TYPE,
+        resource_id=message.id,
+        title=title,
+        url=(
+            "/mail?"
+            f"profileId={quote(message.profile_id, safe='')}"
+            f"&folder={quote(message.folder, safe='')}"
+            f"&uid={quote(message.uid, safe='')}"
+        ),
+        summary=summary or None,
+        body=body or None,
+        keywords=(message.folder[:200], profile.name[:200]),
+        visibility="restricted",
+        acl_tokens=tuple(dict.fromkeys(tokens)),
+        metadata={
+            "profile_id": message.profile_id,
+            "profile_name": profile.name,
+            "folder": message.folder,
+            "uid": message.uid,
+            "date": message.date,
+            "attachment_count": message.attachment_count,
+        },
+        source_revision=updated_at.isoformat(),
+        change_cursor=change_cursor,
+        source_updated_at=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 Mail search source.")
+
+
+def _session(value: object) -> Session:
+    if not isinstance(value, Session):
+        raise TypeError("Mail search requires a SQLAlchemy session.")
+    return value
+
+
+__all__ = [
+    "MailSearchSource",
+    "PROVIDER_ID",
+    "RESOURCE_TYPE",
+    "create_mail_search_source",
+]

tests/test_mailbox_index.py

@@ -14,6 +14,9 @@ class _DeleteQuery:
     def filter(self, *_criteria):
         return self
 
+    def all(self):
+        return []
+
     def delete(self, *, synchronize_session: bool) -> int:
         if synchronize_session is not False:
             raise AssertionError("bulk invalidation must not synchronize loaded cache rows")

tests/test_search_source.py

@@ -0,0 +1,144 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from types import SimpleNamespace
+import unittest
+from unittest.mock import patch
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+from govoplan_access.backend.db.models import Account, User
+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_mail.backend.db.models import (
+    MailMailboxMessageIndex,
+    MailServerProfile,
+)
+from govoplan_mail.backend.search_source import (
+    MailSearchSource,
+    PROVIDER_ID,
+    RESOURCE_TYPE,
+)
+
+
+class MailSearchSourceTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite://")
+        Base.metadata.create_all(
+            self.engine,
+            tables=(
+                Account.__table__,
+                User.__table__,
+                MailServerProfile.__table__,
+                MailMailboxMessageIndex.__table__,
+            ),
+        )
+        self.session = Session(self.engine)
+        self.session.add_all(
+            (
+                Account(
+                    id="account-1",
+                    email="one@example.test",
+                    normalized_email="one@example.test",
+                ),
+                User(
+                    id="user-1",
+                    tenant_id="tenant-1",
+                    account_id="account-1",
+                    email="one@example.test",
+                ),
+                MailServerProfile(
+                    id="profile-1",
+                    tenant_id="tenant-1",
+                    scope_type="user",
+                    scope_id="user-1",
+                    name="Personal mailbox",
+                    slug="personal",
+                    smtp_config={},
+                    imap_config={"host": "imap.example.test"},
+                ),
+                MailMailboxMessageIndex(
+                    id="message-1",
+                    tenant_id="tenant-1",
+                    profile_id="profile-1",
+                    folder="INBOX",
+                    uid="42",
+                    subject="Permit status",
+                    from_header="service@example.test",
+                    body_preview="Your permit is ready",
+                    indexed_at=datetime(2026, 8, 5, tzinfo=timezone.utc),
+                ),
+            )
+        )
+        self.session.commit()
+        self.source = MailSearchSource()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        self.engine.dispose()
+
+    def test_mailbox_backfill_and_profile_recheck_are_bounded(self) -> None:
+        page = self.source.backfill(
+            self.session,
+            request=SearchBackfillRequest(
+                tenant_id="tenant-1",
+                provider_id=PROVIDER_ID,
+                resource_type=RESOURCE_TYPE,
+                rebuild_id="rebuild-1",
+            ),
+        )
+        self.assertEqual(("message-1",), tuple(doc.resource_id for doc in page.documents))
+        self.assertNotIn("password", str(page.documents[0].metadata).casefold())
+        reference = SearchResourceReference(
+            tenant_id="tenant-1",
+            module_id="mail",
+            resource_type=RESOURCE_TYPE,
+            resource_id="message-1",
+        )
+        request = SearchAuthorizationRequest(reference=reference, source_revision="1")
+        with patch(
+            "govoplan_mail.backend.search_source.mail_profile_visible_to_actor",
+            return_value=True,
+        ):
+            allowed = self.source.authorize(
+                self.session,
+                _principal(),
+                requests=(request,),
+            )
+        self.assertTrue(allowed[reference.key])
+        self.assertFalse(
+            self.source.authorize(
+                self.session,
+                _principal(scopes={"mail:mailbox:read"}),
+                requests=(request,),
+            )[reference.key]
+        )
+
+
+def _principal(
+    *,
+    scopes: set[str] | None = None,
+) -> ApiPrincipal:
+    return ApiPrincipal(
+        principal=PrincipalRef(
+            account_id="account-1",
+            membership_id="user-1",
+            tenant_id="tenant-1",
+            scopes=frozenset(
+                scopes or {"mail:mailbox:read", "mail:profile:use"}
+            ),
+        ),
+        account=SimpleNamespace(id="account-1"),
+        user=SimpleNamespace(id="user-1"),
+    )
+
+
+if __name__ == "__main__":
+    unittest.main()