from __future__ import annotations from dataclasses import dataclass, field from datetime import datetime from typing import Any from sqlalchemy import func from govoplan_core.auth import ApiPrincipal from govoplan_core.db.base import utcnow from govoplan_core.core.people import ( PeopleSearchGroup, PersonSearchCandidate, person_selection_key, ) from govoplan_addresses.backend.db.models import AddressBook, AddressList, AddressListEntry, Contact, ContactEmail, ContactPhone from govoplan_addresses.backend.schemas import ContactCreateRequest from govoplan_addresses.backend.service import ( AddressBookError, create_contact, get_visible_address_book, get_visible_address_list, list_address_books, list_address_list_entries, list_address_lists, list_contacts, ) CAPABILITY_ADDRESSES_LOOKUP = "addresses.lookup" CAPABILITY_ADDRESSES_RECIPIENT_SOURCE = "addresses.recipient_source" CAPABILITY_ADDRESSES_CONTACT_WRITER = "addresses.contact_writer" CONTACT_WRITE_OPERATIONS = ("create_contact", "update_contact", "delete_contact") @dataclass(frozen=True, slots=True) class AddressLookupCandidate: contact_id: str address_book_id: str display_name: str email: str | None email_label: str | None = None organization: str | None = None role_title: str | None = None tags: tuple[str, ...] = () source_kind: str = "local" source_ref: str | None = None source_revision: str | None = None provenance: dict[str, Any] = field(default_factory=dict) @dataclass(frozen=True, slots=True) class RecipientSnapshotItem: contact_id: str display_name: str email: str email_label: str | None = None fields: dict[str, Any] = field(default_factory=dict) provenance: dict[str, Any] = field(default_factory=dict) @dataclass(frozen=True, slots=True) class RecipientSourceRef: source_id: str source_label: str source_kind: str source_revision: str recipient_count: int = 0 provenance: dict[str, Any] = field(default_factory=dict) @dataclass(frozen=True, slots=True) class RecipientSourceSnapshot: source_id: str source_label: str source_kind: str source_revision: str generated_at: str recipients: tuple[RecipientSnapshotItem, ...] provenance: dict[str, Any] = field(default_factory=dict) @dataclass(frozen=True, slots=True) class AddressBookWriteDecision: address_book_id: str address_book_label: str | None operation: str allowed: bool reason: str message: str scope_type: str | None = None scope_id: str | None = None tenant_id: str | None = None source_kind: str | None = None read_only: bool = False required_scopes: tuple[str, ...] = () provenance: dict[str, Any] = field(default_factory=dict) @dataclass(frozen=True, slots=True) class ContactWriteResult: contact_id: str address_book_id: str display_name: str email: str | None = None source_kind: str = "local" provenance: dict[str, Any] = field(default_factory=dict) class AddressWriterError(ValueError): def __init__(self, decision: AddressBookWriteDecision): super().__init__(decision.message) self.decision = decision class AddressesLookupCapability: def lookup( self, session: Any, principal: ApiPrincipal, *, query: str, limit: int = 25, address_book_id: str | None = None, ) -> tuple[AddressLookupCandidate, ...]: normalized_limit = max(1, min(limit, 100)) contacts = list_contacts( session, principal, address_book_id=address_book_id, query=query, limit=normalized_limit, ) candidates: list[AddressLookupCandidate] = [] for contact in contacts: email_rows: list[ContactEmail | None] = list(contact.emails) or [None] for email in email_rows: candidates.append(_lookup_candidate(contact, email)) if len(candidates) >= normalized_limit: return tuple(candidates) return tuple(candidates) class AddressesPeopleSearchProvider: """Adapt visible address-book contacts to the shared people-search contract.""" def __init__(self, lookup: AddressesLookupCapability | None = None) -> None: self._lookup = lookup or AddressesLookupCapability() def search_people( self, session: object, principal: object, *, query: str, limit: int = 25, ) -> tuple[PeopleSearchGroup, ...]: if not hasattr(principal, "account_id") or not hasattr(principal, "tenant_id") or not hasattr(principal, "group_ids"): raise AddressBookError("Address contact search requires an authenticated tenant principal.") candidates = self._lookup.lookup( session, principal, # type: ignore[arg-type] - validated principal contract query=query, limit=limit, ) return ( PeopleSearchGroup( key="contacts", label="Contacts", candidates=tuple( PersonSearchCandidate( selection_key=person_selection_key("contact", item.contact_id, email=item.email), kind="contact", reference_id=item.contact_id, display_name=item.display_name, email=item.email, source_module="addresses", source_label="Contacts", source_ref=item.source_ref or f"addresses:contact:{item.contact_id}", source_revision=item.source_revision, description=" ยท ".join(part for part in (item.organization, item.role_title) if part) or None, provenance=dict(item.provenance), metadata={ "address_book_id": item.address_book_id, "email_label": item.email_label, "organization": item.organization, "role_title": item.role_title, "tags": tuple(item.tags), "source_kind": item.source_kind, }, ) for item in candidates ), ), ) class AddressesContactWriterCapability: def list_write_targets( self, session: Any, principal: ApiPrincipal, *, operation: str = "create_contact", ) -> tuple[AddressBookWriteDecision, ...]: books = list_address_books(session, principal) return tuple(_address_book_write_decision(principal, book, operation=operation) for book in books) def can_write_to_address_book( self, session: Any, principal: ApiPrincipal, *, address_book_id: str, operation: str = "create_contact", ) -> AddressBookWriteDecision: try: book = get_visible_address_book(session, principal, address_book_id, include_deleted=True) except AddressBookError: return _blocked_write_decision( address_book_id=address_book_id, operation=operation, reason="not_visible", message="Address book is not visible to the current principal.", ) return _address_book_write_decision(principal, book, operation=operation) def create_contact( self, session: Any, principal: ApiPrincipal, *, address_book_id: str, payload: ContactCreateRequest | dict[str, Any], provenance: dict[str, Any] | None = None, ) -> ContactWriteResult: decision = self.can_write_to_address_book(session, principal, address_book_id=address_book_id, operation="create_contact") if not decision.allowed: raise AddressWriterError(decision) request = payload if isinstance(payload, ContactCreateRequest) else ContactCreateRequest.model_validate(payload) if provenance: request = request.model_copy(update={"provenance": {**request.provenance, **provenance}}) contact = create_contact(session, principal, address_book_id, request) return _contact_write_result(contact) class AddressesRecipientSourceCapability: def list_sources( self, session: Any, principal: ApiPrincipal, ) -> tuple[RecipientSourceRef, ...]: from govoplan_addresses.backend.service import address_book_contact_counts, address_list_entry_counts books = list_address_books(session, principal) counts = address_book_contact_counts(session, [book.id for book in books]) contact_updated_at = _address_book_contact_updated_at(session, [book.id for book in books]) book_sources = tuple( _recipient_source_ref( book, recipient_count=counts.get(book.id, 0), latest_contact_updated_at=contact_updated_at.get(book.id), ) for book in books ) address_lists = list_address_lists(session, principal) address_list_counts = address_list_entry_counts(session, [address_list.id for address_list in address_lists]) address_list_updated_at = _address_list_updated_at(session, [address_list.id for address_list in address_lists]) list_sources = tuple( _address_list_source_ref( address_list, recipient_count=address_list_counts.get(address_list.id, 0), latest_entry_updated_at=address_list_updated_at.get(address_list.id), ) for address_list in address_lists ) return book_sources + list_sources def snapshot_address_book( self, session: Any, principal: ApiPrincipal, *, address_book_id: str, ) -> RecipientSourceSnapshot: book = get_visible_address_book(session, principal, address_book_id) contacts = _active_address_book_contacts(session, book.id) recipients: list[RecipientSnapshotItem] = [] for contact in contacts: for email in contact.emails: recipients.append(_recipient_snapshot_item(contact, email)) revision = _recipient_source_revision(book, _address_book_contact_updated_at(session, [book.id]).get(book.id)) return RecipientSourceSnapshot( source_id=f"addresses:address_book:{book.id}", source_label=book.name, source_kind=book.source_kind, source_revision=revision, generated_at=utcnow().isoformat(), recipients=tuple(recipients), provenance={ "module": "addresses", "address_book_id": book.id, "scope_type": book.scope_type, "scope_id": book.scope_id, "tenant_id": book.tenant_id, }, ) def snapshot_address_list( self, session: Any, principal: ApiPrincipal, *, address_list_id: str, ) -> RecipientSourceSnapshot: address_list = get_visible_address_list(session, principal, address_list_id) entries = list_address_list_entries(session, principal, address_list.id) recipients: list[RecipientSnapshotItem] = [] for entry in entries: email = _entry_email(entry) if email is None: continue recipients.append(_recipient_snapshot_item(entry.contact, email, address_list=address_list, address_list_entry=entry)) revision = _address_list_source_revision(address_list, _address_list_updated_at(session, [address_list.id]).get(address_list.id)) return RecipientSourceSnapshot( source_id=f"addresses:address_list:{address_list.id}", source_label=address_list.name, source_kind="address_list", source_revision=revision, generated_at=utcnow().isoformat(), recipients=tuple(recipients), provenance={ "module": "addresses", "address_book_id": address_list.address_book_id, "address_list_id": address_list.id, "scope_type": address_list.address_book.scope_type, "scope_id": address_list.address_book.scope_id, "tenant_id": address_list.tenant_id, }, ) def snapshot( self, session: Any, principal: ApiPrincipal, *, source_id: str, ) -> RecipientSourceSnapshot: book_prefix = "addresses:address_book:" if source_id.startswith(book_prefix): return self.snapshot_address_book(session, principal, address_book_id=source_id.removeprefix(book_prefix)) list_prefix = "addresses:address_list:" if source_id.startswith(list_prefix): return self.snapshot_address_list(session, principal, address_list_id=source_id.removeprefix(list_prefix)) raise ValueError(f"Unsupported addresses recipient source id: {source_id}") def lookup_capability(_context: Any) -> AddressesLookupCapability: return AddressesLookupCapability() def people_search_capability(_context: Any) -> AddressesPeopleSearchProvider: return AddressesPeopleSearchProvider() def recipient_source_capability(_context: Any) -> AddressesRecipientSourceCapability: return AddressesRecipientSourceCapability() def contact_writer_capability(_context: Any) -> AddressesContactWriterCapability: return AddressesContactWriterCapability() def _operation_required_scopes(operation: str) -> tuple[str, ...] | None: if operation in ("create_contact", "update_contact"): return ("addresses:address_book:read", "addresses:contact:write") if operation == "delete_contact": return ("addresses:address_book:read", "addresses:contact:delete") return None def _address_book_write_decision(principal: ApiPrincipal, book: AddressBook, *, operation: str) -> AddressBookWriteDecision: required_scopes = _operation_required_scopes(operation) if required_scopes is None: return _blocked_write_decision( address_book_id=book.id, operation=operation, reason="unsupported_operation", message=f"Unsupported address-book write operation: {operation}", book=book, ) missing_scopes = tuple(scope for scope in required_scopes if not principal.has(scope)) if missing_scopes: return _blocked_write_decision( address_book_id=book.id, operation=operation, reason="missing_scope", message=f"Missing scope: {missing_scopes[0]}", book=book, required_scopes=required_scopes, ) if book.deleted_at is not None: return _blocked_write_decision( address_book_id=book.id, operation=operation, reason="address_book_deleted", message="Address book is deleted.", book=book, required_scopes=required_scopes, ) if book.read_only: return _blocked_write_decision( address_book_id=book.id, operation=operation, reason="address_book_read_only", message="Address book is read-only.", book=book, required_scopes=required_scopes, ) return AddressBookWriteDecision( address_book_id=book.id, address_book_label=book.name, operation=operation, allowed=True, reason="allowed", message="Address book accepts this write operation.", scope_type=book.scope_type, scope_id=book.scope_id, tenant_id=book.tenant_id, source_kind=book.source_kind, read_only=book.read_only, required_scopes=required_scopes, provenance=_address_book_provenance(book), ) def _blocked_write_decision( *, address_book_id: str, operation: str, reason: str, message: str, book: AddressBook | None = None, required_scopes: tuple[str, ...] = (), ) -> AddressBookWriteDecision: return AddressBookWriteDecision( address_book_id=address_book_id, address_book_label=book.name if book else None, operation=operation, allowed=False, reason=reason, message=message, scope_type=book.scope_type if book else None, scope_id=book.scope_id if book else None, tenant_id=book.tenant_id if book else None, source_kind=book.source_kind if book else None, read_only=book.read_only if book else False, required_scopes=required_scopes, provenance=_address_book_provenance(book) if book else {"module": "addresses", "address_book_id": address_book_id}, ) def _lookup_candidate(contact: Contact, email: ContactEmail | None) -> AddressLookupCandidate: return AddressLookupCandidate( contact_id=contact.id, address_book_id=contact.address_book_id, display_name=contact.display_name, email=email.email if email is not None else None, email_label=email.label if email is not None else None, organization=contact.organization, role_title=contact.role_title, tags=tuple(contact.tags or ()), source_kind=contact.source_kind, source_ref=contact.source_ref, source_revision=contact.source_revision, provenance=_contact_provenance(contact), ) def _recipient_snapshot_item( contact: Contact, email: ContactEmail, *, address_list: AddressList | None = None, address_list_entry: AddressListEntry | None = None, ) -> RecipientSnapshotItem: primary_phone = next((phone.phone for phone in contact.phones if phone.is_primary), contact.phones[0].phone if contact.phones else None) provenance = { **_contact_provenance(contact), "email_id": email.id, "email_label": email.label, "email_primary": email.is_primary, } if address_list is not None and address_list_entry is not None: provenance.update( { "address_list_id": address_list.id, "address_list_entry_id": address_list_entry.id, "address_list_name": address_list.name, "address_list_entry_kind": address_list_entry.target_kind, } ) return RecipientSnapshotItem( contact_id=contact.id, display_name=contact.display_name, email=email.email, email_label=email.label, fields={ "given_name": contact.given_name, "family_name": contact.family_name, "organization": contact.organization, "role_title": contact.role_title, "phone": primary_phone, "tags": list(contact.tags or ()), }, provenance=provenance, ) def _contact_provenance(contact: Contact) -> dict[str, Any]: return { "module": "addresses", "contact_id": contact.id, "address_book_id": contact.address_book_id, "source_kind": contact.source_kind, "source_ref": contact.source_ref, "source_revision": contact.source_revision, } def _address_book_provenance(book: AddressBook) -> dict[str, Any]: return { "module": "addresses", "address_book_id": book.id, "scope_type": book.scope_type, "scope_id": book.scope_id, "tenant_id": book.tenant_id, "source_kind": book.source_kind, "source_ref": book.source_ref, } def _entry_email(entry: AddressListEntry) -> ContactEmail | None: if entry.contact_postal_address_id is not None: return None if entry.contact_email is not None: return entry.contact_email return next((email for email in entry.contact.emails if email.is_primary), entry.contact.emails[0] if entry.contact.emails else None) def _contact_write_result(contact: Contact) -> ContactWriteResult: primary_email = next((email.email for email in contact.emails if email.is_primary), contact.emails[0].email if contact.emails else None) return ContactWriteResult( contact_id=contact.id, address_book_id=contact.address_book_id, display_name=contact.display_name, email=primary_email, source_kind=contact.source_kind, provenance={ **_contact_provenance(contact), "contact_provenance": contact.provenance, }, ) def _recipient_source_ref(book: AddressBook, *, recipient_count: int, latest_contact_updated_at: datetime | None = None) -> RecipientSourceRef: return RecipientSourceRef( source_id=f"addresses:address_book:{book.id}", source_label=book.name, source_kind=book.source_kind, source_revision=_recipient_source_revision(book, latest_contact_updated_at), recipient_count=recipient_count, provenance={ "module": "addresses", "address_book_id": book.id, "scope_type": book.scope_type, "scope_id": book.scope_id, "tenant_id": book.tenant_id, }, ) def _address_list_source_ref(address_list: AddressList, *, recipient_count: int, latest_entry_updated_at: datetime | None = None) -> RecipientSourceRef: return RecipientSourceRef( source_id=f"addresses:address_list:{address_list.id}", source_label=address_list.name, source_kind="address_list", source_revision=_address_list_source_revision(address_list, latest_entry_updated_at), recipient_count=recipient_count, provenance={ "module": "addresses", "address_book_id": address_list.address_book_id, "address_list_id": address_list.id, "scope_type": address_list.address_book.scope_type, "scope_id": address_list.address_book.scope_id, "tenant_id": address_list.tenant_id, }, ) def _active_address_book_contacts(session: Any, address_book_id: str) -> list[Contact]: return ( session.query(Contact) .filter(Contact.address_book_id == address_book_id, Contact.deleted_at.is_(None)) .order_by(Contact.display_name.asc(), Contact.id.asc()) .all() ) def _address_book_contact_updated_at(session: Any, address_book_ids: list[str]) -> dict[str, datetime]: if not address_book_ids: return {} revisions: dict[str, datetime] = {} contact_rows = ( session.query(Contact.address_book_id, func.max(Contact.updated_at)) .filter(Contact.address_book_id.in_(address_book_ids), Contact.deleted_at.is_(None)) .group_by(Contact.address_book_id) .all() ) deletion_rows = ( session.query(Contact.address_book_id, func.max(Contact.deleted_at)) .filter(Contact.address_book_id.in_(address_book_ids), Contact.deleted_at.is_not(None)) .group_by(Contact.address_book_id) .all() ) email_rows = ( session.query(Contact.address_book_id, func.max(ContactEmail.updated_at)) .join(Contact, ContactEmail.contact_id == Contact.id) .filter(Contact.address_book_id.in_(address_book_ids), Contact.deleted_at.is_(None)) .group_by(Contact.address_book_id) .all() ) phone_rows = ( session.query(Contact.address_book_id, func.max(ContactPhone.updated_at)) .join(Contact, ContactPhone.contact_id == Contact.id) .filter(Contact.address_book_id.in_(address_book_ids), Contact.deleted_at.is_(None)) .group_by(Contact.address_book_id) .all() ) for address_book_id, updated_at in [*contact_rows, *deletion_rows, *email_rows, *phone_rows]: if updated_at is None: continue key = str(address_book_id) revisions[key] = max(revisions.get(key, updated_at), updated_at) return revisions def _address_list_updated_at(session: Any, address_list_ids: list[str]) -> dict[str, datetime]: if not address_list_ids: return {} revisions: dict[str, datetime] = {} entry_rows = ( session.query(AddressListEntry.address_list_id, func.max(AddressListEntry.updated_at)) .filter(AddressListEntry.address_list_id.in_(address_list_ids)) .group_by(AddressListEntry.address_list_id) .all() ) contact_rows = ( session.query(AddressListEntry.address_list_id, func.max(Contact.updated_at)) .join(Contact, AddressListEntry.contact_id == Contact.id) .filter(AddressListEntry.address_list_id.in_(address_list_ids), Contact.deleted_at.is_(None)) .group_by(AddressListEntry.address_list_id) .all() ) email_rows = ( session.query(AddressListEntry.address_list_id, func.max(ContactEmail.updated_at)) .join(ContactEmail, AddressListEntry.contact_email_id == ContactEmail.id) .filter(AddressListEntry.address_list_id.in_(address_list_ids)) .group_by(AddressListEntry.address_list_id) .all() ) for address_list_id, updated_at in [*entry_rows, *contact_rows, *email_rows]: if updated_at is None: continue key = str(address_list_id) revisions[key] = max(revisions.get(key, updated_at), updated_at) return revisions def _recipient_source_revision(book: AddressBook, latest_contact_updated_at: datetime | None = None) -> str: stamps: list[datetime] = [book.updated_at] if latest_contact_updated_at is not None: stamps.append(latest_contact_updated_at) return max(stamps).isoformat() def _address_list_source_revision(address_list: AddressList, latest_entry_updated_at: datetime | None = None) -> str: stamps: list[datetime] = [address_list.updated_at] if latest_entry_updated_at is not None: stamps.append(latest_entry_updated_at) return max(stamps).isoformat()