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", ]