from __future__ import annotations from dataclasses import asdict, dataclass, field, replace from datetime import UTC, datetime import hashlib import json from typing import Any from sqlalchemy import func, or_ from sqlalchemy.orm import selectinload from govoplan_core.auth import ApiPrincipal from govoplan_core.core.distribution_lists import ( DistributionChannelCandidate, DistributionExplanation, DistributionSourceReference, RecipientChannelFacts, RecipientChannelFactsRequest, ) from govoplan_core.core.contact_points import ( CONTACT_POINT_CONTRACT_VERSION, ContactPointCandidate, ContactPointResolution, ContactPointResolutionProvider, ContactPointResolutionRequest, ContactPointSnapshotRef, ContactPointSourcePreview, ContactPointSourceRequest, ) 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, ContactChannelRule, ContactEmail, ContactPhone, ContactPointQualityDecision, ContactPointSnapshot, ContactPostalAddress, ) from govoplan_addresses.backend.schemas import ContactCreateRequest from govoplan_addresses.backend.service import ( AddressBookError, create_contact, current_contact_quality, get_visible_address_book, get_visible_address_list, get_visible_contact, list_address_books, list_address_list_entries, list_address_lists, list_contacts, resolve_contact_redirect, ) 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 RecipientSnapshotExcludedItem: contact_id: str display_name: str channel: str target: str contact_point_id: str | None = None status: str = "suppressed" reason_code: str | None = None explanation: str | None = None 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, ...] excluded: tuple[RecipientSnapshotExcludedItem, ...] = () purpose: str | None = None requested_channels: tuple[str, ...] = () 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, purpose: str | None = None, requested_channels: tuple[str, ...] = ("email",), ) -> RecipientSourceSnapshot: book = get_visible_address_book(session, principal, address_book_id) contacts = _active_address_book_contacts(session, book.id) recipients: list[RecipientSnapshotItem] = [] excluded: list[RecipientSnapshotExcludedItem] = [] if purpose is None: for contact in contacts: for email in contact.emails: recipients.append(_recipient_snapshot_item(contact, email)) else: for contact in contacts: included_rows, excluded_rows = _governed_email_snapshot_rows( session, principal, contact, purpose=purpose, ) recipients.extend(included_rows) excluded.extend(excluded_rows) 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), excluded=tuple(excluded), purpose=purpose, requested_channels=requested_channels, provenance={ "module": "addresses", "address_book_id": book.id, "scope_type": book.scope_type, "scope_id": book.scope_id, "tenant_id": book.tenant_id, "governance_applied": purpose is not None, "included_count": len(recipients), "excluded_count": len(excluded), }, ) def snapshot_address_list( self, session: Any, principal: ApiPrincipal, *, address_list_id: str, purpose: str | None = None, requested_channels: tuple[str, ...] = ("email",), ) -> 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] = [] excluded: list[RecipientSnapshotExcludedItem] = [] for entry in entries: email = _entry_email(entry) if email is None: continue if purpose is None: recipients.append(_recipient_snapshot_item(entry.contact, email, address_list=address_list, address_list_entry=entry)) continue included_rows, excluded_rows = _governed_email_snapshot_rows( session, principal, entry.contact, purpose=purpose, email_ids={email.id}, address_list=address_list, address_list_entry=entry, ) recipients.extend(included_rows) excluded.extend(excluded_rows) 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), excluded=tuple(excluded), purpose=purpose, requested_channels=requested_channels, 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, "governance_applied": purpose is not None, "included_count": len(recipients), "excluded_count": len(excluded), }, ) def snapshot( self, session: Any, principal: ApiPrincipal, *, source_id: str, purpose: str | None = None, requested_channels: tuple[str, ...] = ("email",), ) -> 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), purpose=purpose, requested_channels=requested_channels, ) 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), purpose=purpose, requested_channels=requested_channels, ) raise ValueError(f"Unsupported addresses recipient source id: {source_id}") class AddressesChannelFactsCapability: """Resolve effective contact-point facts without making a Policy decision.""" def resolve_channel_facts( self, session: Any, principal: ApiPrincipal, *, request: RecipientChannelFactsRequest, ) -> RecipientChannelFacts: if request.tenant_id != principal.tenant_id: raise PermissionError("Channel facts cannot be resolved across tenants.") contact_id = _channel_facts_contact_id(request.source) contact = get_visible_contact(session, principal, contact_id) effective_at = _aware_datetime(request.effective_at) active_rules = [ rule for rule in contact.channel_rules if _rule_matches_purpose(rule, request.purpose) and _rule_is_effective(rule, effective_at) ] expired_rules = [ rule for rule in contact.channel_rules if _rule_matches_purpose(rule, request.purpose) and rule.effective_until is not None and _aware_datetime(rule.effective_until) <= effective_at ] revision, fingerprint = _contact_channel_revision(contact) quality = current_contact_quality(contact, effective_at=effective_at) source = DistributionSourceReference( provider="addresses", resource_type="contact", resource_id=contact.id, revision=revision, fingerprint=fingerprint, label=contact.display_name, metadata={ "address_book_id": contact.address_book_id, "source_kind": contact.source_kind, "source_ref": contact.source_ref, }, ) requested_channels = set(request.requested_channels) candidates: list[DistributionChannelCandidate] = [] for channel, point_id, target, target_key, intrinsically_valid in _contact_channel_points(contact): if requested_channels and channel not in requested_channels: continue matching = [ rule for rule in active_rules if rule.channel == channel and (rule.contact_point_id is None or rule.contact_point_id == point_id) ] selected = max(matching, key=_channel_rule_priority, default=None) state = selected.decision if selected is not None else "unknown" status = _channel_candidate_status(state, intrinsically_valid=intrinsically_valid) reason_code = ( f"addresses.channel.{state}" if intrinsically_valid else f"addresses.{channel}.invalid" ) explanation = ( _channel_rule_explanation(selected) if selected is not None else ( f"The {channel.replace('_', ' ')} contact point has no applicable governance fact." if intrinsically_valid else f"The {channel.replace('_', ' ')} contact point is incomplete or invalid." ) ) quality_decision = quality.get((channel, point_id)) or quality.get( (channel, None) ) if quality_decision is not None and quality_decision.state != "valid": status = ( "stale" if quality_decision.state == "stale" else "invalid" ) reason_code = quality_decision.reason_code explanation = quality_decision.reason or ( f"This {channel.replace('_', ' ')} contact point is marked " f"{quality_decision.state}." ) candidates.append( DistributionChannelCandidate( channel=channel, target=target, target_key=target_key, status=status, contact_point_id=point_id, locale=selected.locale if selected is not None else None, preferred=any(rule.decision == "preferred" for rule in matching), reason_code=reason_code, explanation=explanation, source=source, decision_provenance={ "provider": "addresses", "governance_state": state, "effective_at": effective_at.isoformat(), "purpose": request.purpose, "selected_rule_id": selected.id if selected is not None else None, "rule_ids": [rule.id for rule in matching], "legal_basis": selected.legal_basis if selected is not None else None, "evidence_ref": selected.evidence_ref if selected is not None else None, "preference_rank": selected.preference_rank if selected is not None else None, "quality_decision_id": ( quality_decision.id if quality_decision is not None else None ), "quality_state": ( quality_decision.state if quality_decision is not None else "valid" ), "quality_evidence_ref": ( quality_decision.evidence_ref if quality_decision is not None else None ), }, ) ) explanations = tuple( DistributionExplanation( code="addresses.channel_fact.expired", message=( f"A {rule.channel.replace('_', ' ')} governance fact expired and was not applied." ), severity="info", provider="addresses", source=source, provenance={"rule_id": rule.id, "decision": rule.decision}, ) for rule in expired_rules ) return RecipientChannelFacts( candidates=tuple(candidates), explanations=explanations, source_revision=revision, source_fingerprint=fingerprint, provenance={ "module": "addresses", "contact_id": contact.id, "address_book_id": contact.address_book_id, "effective_at": effective_at.isoformat(), "purpose": request.purpose, "active_rule_ids": [rule.id for rule in active_rules], "expired_rule_ids": [rule.id for rule in expired_rules], "source_revision": revision, "source_fingerprint": fingerprint, }, ) @dataclass(frozen=True, slots=True) class _SourceContactContext: contact: Contact allowed_contact_point_ids: frozenset[str] | None = None address_list_entry_ids: tuple[str, ...] = () class AddressesContactPointResolutionCapability(ContactPointResolutionProvider): """Resolve and freeze channel-neutral contact points without policy imports.""" def resolve_contact_points( self, session: Any, principal: ApiPrincipal, *, request: ContactPointResolutionRequest, ) -> ContactPointResolution: self._assert_tenant(principal, request.tenant_id) contacts = _contacts_for_subject(session, principal, request.subject) if not contacts: return _unresolved_contact_point_resolution( request, code="addresses.subject.unresolved", message="No visible Addresses contact is linked to this subject reference.", ) if len(contacts) > 1: return _unresolved_contact_point_resolution( request, code="addresses.subject.ambiguous", message="More than one visible Addresses contact is linked to this subject reference.", status="ambiguous", provenance={"contact_ids": [item.id for item in contacts]}, ) return _resolve_contact_points_for_contact( session, principal, request=request, context=_SourceContactContext(contact=contacts[0]), ) def preview_source( self, session: Any, principal: ApiPrincipal, *, request: ContactPointSourceRequest, offset: int = 0, limit: int = 100, ) -> ContactPointSourcePreview: self._assert_tenant(principal, request.tenant_id) bounded_offset = max(0, offset) bounded_limit = max(1, min(limit, 500)) source, source_revision, contexts = _contact_point_source_contexts( session, principal, request.source_id, ) if len(contexts) > max(1, min(request.max_items, 20_000)): raise ValueError( f"Contact-point source exceeds the configured {request.max_items} item limit." ) selected = contexts[bounded_offset : bounded_offset + bounded_limit] resolutions = tuple( _resolve_contact_points_for_contact( session, principal, request=_contact_point_resolution_request(request, item.contact), context=item, ) for item in selected ) source_fingerprint = _source_fingerprint(request.source_id, source_revision) return ContactPointSourcePreview( contract_version=CONTACT_POINT_CONTRACT_VERSION, source=source, request=request, resolutions=resolutions, total_count=len(contexts), usable_count=sum(len(item.candidates) for item in resolutions), excluded_count=sum( len(item.excluded) + int(not item.candidates and not item.excluded) for item in resolutions ), offset=bounded_offset, limit=bounded_limit, has_more=bounded_offset + len(resolutions) < len(contexts), source_revision=source_revision, source_fingerprint=source_fingerprint, generated_at=utcnow(), provenance={ "module": "addresses", "contract_version": CONTACT_POINT_CONTRACT_VERSION, "bounded": True, }, ) def freeze_source( self, session: Any, principal: ApiPrincipal, *, request: ContactPointSourceRequest, ) -> ContactPointSnapshotRef: self._assert_tenant(principal, request.tenant_id) source, source_revision, contexts = _contact_point_source_contexts( session, principal, request.source_id, ) max_items = max(1, min(request.max_items, 20_000)) if len(contexts) > max_items: raise ValueError( f"Contact-point source exceeds the configured {request.max_items} item limit." ) resolutions = tuple( _resolve_contact_points_for_contact( session, principal, request=_contact_point_resolution_request(request, item.contact), context=item, ) for item in contexts ) generated_at = utcnow() source_fingerprint = _source_fingerprint(request.source_id, source_revision) request_payload = _json_value(asdict(request)) resolution_payload = [_json_value(asdict(item)) for item in resolutions] snapshot_hash = hashlib.sha256( json.dumps( { "contract_version": CONTACT_POINT_CONTRACT_VERSION, "source": _json_value(asdict(source)), "request": request_payload, "resolutions": resolution_payload, "source_revision": source_revision, "source_fingerprint": source_fingerprint, }, sort_keys=True, separators=(",", ":"), ).encode("utf-8") ).hexdigest() snapshot = ContactPointSnapshot( tenant_id=request.tenant_id, source_id=request.source_id, contract_version=CONTACT_POINT_CONTRACT_VERSION, source_revision=source_revision, source_fingerprint=source_fingerprint, purpose=request.purpose, effective_at=request.effective_at, generated_at=generated_at, request_payload=request_payload, resolution_payload=resolution_payload, recipient_count=sum(len(item.candidates) for item in resolutions), excluded_count=sum( len(item.excluded) + int(not item.candidates and not item.excluded) for item in resolutions ), snapshot_hash=snapshot_hash, created_by_account_id=principal.account_id, provenance={ "module": "addresses", "source": _json_value(asdict(source)), "immutable": True, }, ) session.add(snapshot) session.flush() return _contact_point_snapshot_ref(snapshot) def get_snapshot( self, session: Any, principal: ApiPrincipal, *, snapshot_id: str, ) -> ContactPointSnapshotRef | None: snapshot = ( session.query(ContactPointSnapshot) .filter( ContactPointSnapshot.id == snapshot_id, ContactPointSnapshot.tenant_id == principal.tenant_id, ) .one_or_none() ) return _contact_point_snapshot_ref(snapshot) if snapshot is not None else None @staticmethod def _assert_tenant(principal: ApiPrincipal, tenant_id: str) -> None: if tenant_id != principal.tenant_id: raise PermissionError("Contact points cannot be resolved across tenants.") 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 channel_facts_capability(_context: Any) -> AddressesChannelFactsCapability: return AddressesChannelFactsCapability() def contact_point_resolution_capability( _context: Any, ) -> AddressesContactPointResolutionCapability: return AddressesContactPointResolutionCapability() 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, channel_decision: dict[str, Any] | 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, } ) if channel_decision is not None: provenance["channel_decision"] = dict(channel_decision) 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 _governed_email_snapshot_rows( session: Any, principal: ApiPrincipal, contact: Contact, *, purpose: str, email_ids: set[str] | None = None, address_list: AddressList | None = None, address_list_entry: AddressListEntry | None = None, ) -> tuple[list[RecipientSnapshotItem], list[RecipientSnapshotExcludedItem]]: effective_at = utcnow() facts = AddressesChannelFactsCapability().resolve_channel_facts( session, principal, request=RecipientChannelFactsRequest( tenant_id=principal.tenant_id, source=DistributionSourceReference( provider="addresses", resource_type="contact", resource_id=contact.id, ), recipient_key=f"contact:{contact.id}", effective_at=effective_at, purpose=purpose, requested_channels=("email",), ), ) emails_by_id = {item.id: item for item in contact.emails} included: list[RecipientSnapshotItem] = [] excluded: list[RecipientSnapshotExcludedItem] = [] for candidate in facts.candidates: if candidate.channel != "email": continue point_id = candidate.contact_point_id if email_ids is not None and point_id not in email_ids: continue email = emails_by_id.get(point_id or "") if email is None: continue decision = { **dict(candidate.decision_provenance), "status": candidate.status, "reason_code": candidate.reason_code, "explanation": candidate.explanation, "source_revision": facts.source_revision, "source_fingerprint": facts.source_fingerprint, } if candidate.status in {"usable", "stale"}: included.append( _recipient_snapshot_item( contact, email, address_list=address_list, address_list_entry=address_list_entry, channel_decision=decision, ) ) else: provenance = { **_contact_provenance(contact), "channel_decision": decision, } 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, } ) excluded.append( RecipientSnapshotExcludedItem( contact_id=contact.id, display_name=contact.display_name, channel="email", target=email.email, contact_point_id=email.id, status=candidate.status, reason_code=candidate.reason_code, explanation=candidate.explanation, provenance=provenance, ) ) return included, excluded 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) .options( selectinload(Contact.emails), selectinload(Contact.phones), selectinload(Contact.postal_addresses), selectinload(Contact.channel_rules), ) .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() ) postal_rows = ( session.query(Contact.address_book_id, func.max(ContactPostalAddress.updated_at)) .join(Contact, ContactPostalAddress.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() ) rule_rows = ( session.query(Contact.address_book_id, func.max(ContactChannelRule.updated_at)) .join(Contact, ContactChannelRule.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() ) quality_rows = ( session.query( Contact.address_book_id, func.max(ContactPointQualityDecision.updated_at), ) .join(Contact, ContactPointQualityDecision.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, *postal_rows, *rule_rows, *quality_rows, ]: if updated_at is None: continue key = str(address_book_id) revisions[key] = max( _aware_datetime(revisions.get(key, updated_at)), _aware_datetime(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(Contact, AddressListEntry.contact_id == Contact.id) .join(ContactEmail, ContactEmail.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() ) rule_rows = ( session.query(AddressListEntry.address_list_id, func.max(ContactChannelRule.updated_at)) .join(ContactChannelRule, AddressListEntry.contact_id == ContactChannelRule.contact_id) .filter(AddressListEntry.address_list_id.in_(address_list_ids)) .group_by(AddressListEntry.address_list_id) .all() ) quality_rows = ( session.query( AddressListEntry.address_list_id, func.max(ContactPointQualityDecision.updated_at), ) .join(ContactPointQualityDecision, AddressListEntry.contact_id == ContactPointQualityDecision.contact_id) .filter(AddressListEntry.address_list_id.in_(address_list_ids)) .group_by(AddressListEntry.address_list_id) .all() ) postal_rows = ( session.query( AddressListEntry.address_list_id, func.max(ContactPostalAddress.updated_at), ) .join(Contact, AddressListEntry.contact_id == Contact.id) .join(ContactPostalAddress, ContactPostalAddress.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() ) for address_list_id, updated_at in [ *entry_rows, *contact_rows, *email_rows, *postal_rows, *rule_rows, *quality_rows, ]: if updated_at is None: continue key = str(address_list_id) revisions[key] = max( _aware_datetime(revisions.get(key, updated_at)), _aware_datetime(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(_aware_datetime(item) for item in 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(_aware_datetime(item) for item in stamps).isoformat() def _channel_facts_contact_id(source: DistributionSourceReference) -> str: metadata_contact_id = source.metadata.get("contact_id") if metadata_contact_id: return str(metadata_contact_id) if source.resource_type not in {"contact", "address_contact", "address_email"}: raise ValueError(f"Unsupported Addresses channel-facts source: {source.resource_type}.") return source.resource_id def _contact_channel_points( contact: Contact, ) -> tuple[tuple[str, str | None, str, str, bool], ...]: points: list[tuple[str, str | None, str, str, bool]] = [] for email in contact.emails: target = email.email.strip() valid = bool(target and "@" in target and len(target) <= 320) points.append(("email", email.id, target, f"email:{target.casefold()}", valid)) for address in contact.postal_addresses: target = ", ".join( item for item in ( address.street, " ".join(part for part in (address.postal_code, address.locality) if part), address.region, address.country, ) if item ) points.append(("postal", address.id, target, f"postal:{' '.join(target.casefold().split())}", bool(target))) provenance = dict(contact.provenance or {}) internal_target = provenance.get("internal_mail_account_id") or provenance.get("account_id") if internal_target: target = str(internal_target).strip() points.append(("internal_mail", None, target, f"internal_mail:{target.casefold()}", bool(target))) portal_target = provenance.get("portal_target") or provenance.get("portal_account_id") if portal_target: target = str(portal_target).strip() points.append(("portal", None, target, f"portal:{target.casefold()}", bool(target))) return tuple(points) def _rule_matches_purpose(rule: ContactChannelRule, purpose: str | None) -> bool: return rule.purpose is None or (purpose is not None and rule.purpose == purpose) def _rule_is_effective(rule: ContactChannelRule, effective_at: datetime) -> bool: return ( (rule.effective_from is None or _aware_datetime(rule.effective_from) <= effective_at) and (rule.effective_until is None or _aware_datetime(rule.effective_until) > effective_at) ) def _channel_rule_priority(rule: ContactChannelRule) -> tuple[int, int, int, datetime]: decision_priority = { "invalid": 90, "returned": 80, "suppressed": 70, "opted_out": 60, "temporarily_unavailable": 50, "preferred": 30, "opted_in": 20, "allowed": 10, }.get(rule.decision, 0) return ( decision_priority, int(rule.contact_point_id is not None), int(rule.purpose is not None), _aware_datetime(rule.updated_at), ) def _channel_candidate_status(state: str, *, intrinsically_valid: bool) -> str: if not intrinsically_valid or state in {"invalid", "returned"}: return "invalid" if state in {"suppressed", "opted_out", "temporarily_unavailable"}: return "suppressed" return "usable" def _channel_rule_explanation(rule: ContactChannelRule) -> str: if rule.reason: return rule.reason return { "allowed": "Use of this contact point is explicitly allowed.", "opted_in": "The contact opted in to this communication channel.", "preferred": "This is a preferred communication channel.", "opted_out": "The contact opted out of this communication channel.", "suppressed": "This contact point is suppressed.", "invalid": "This contact point is marked invalid.", "returned": "Delivery to this contact point was returned.", "temporarily_unavailable": "This contact point is temporarily unavailable.", }.get(rule.decision, "An Addresses governance fact was applied.") def _contact_channel_revision(contact: Contact) -> tuple[str, str]: stamps = [contact.updated_at] stamps.extend( item.updated_at for item in ( *contact.emails, *contact.postal_addresses, *contact.channel_rules, *contact.quality_decisions, ) ) revision = max(_aware_datetime(item) for item in stamps).isoformat() payload = { "contact_id": contact.id, "revision": revision, "emails": [ {"id": item.id, "email": item.email, "primary": item.is_primary} for item in contact.emails ], "postal": [ { "id": item.id, "street": item.street, "postal_code": item.postal_code, "locality": item.locality, "region": item.region, "country": item.country, "primary": item.is_primary, } for item in contact.postal_addresses ], "rules": [ { "id": rule.id, "channel": rule.channel, "purpose": rule.purpose, "point": rule.contact_point_id, "decision": rule.decision, "basis": rule.legal_basis, "evidence": rule.evidence_ref, "rank": rule.preference_rank, "locale": rule.locale, "from": rule.effective_from.isoformat() if rule.effective_from else None, "until": rule.effective_until.isoformat() if rule.effective_until else None, "updated": rule.updated_at.isoformat(), } for rule in sorted(contact.channel_rules, key=lambda item: item.id) ], "quality": [ { "id": item.id, "channel": item.channel, "point": item.contact_point_id, "state": item.state, "reason_code": item.reason_code, "evidence": item.evidence_ref, "from": item.effective_from.isoformat(), "until": item.effective_until.isoformat() if item.effective_until else None, "updated": item.updated_at.isoformat(), } for item in sorted(contact.quality_decisions, key=lambda row: row.id) ], } fingerprint = hashlib.sha256( json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8") ).hexdigest() return revision, fingerprint def _aware_datetime(value: datetime) -> datetime: if value.tzinfo is None: return value.replace(tzinfo=UTC) return value def _contacts_for_subject( session: Any, principal: ApiPrincipal, subject: DistributionSourceReference, ) -> list[Contact]: direct_id = subject.metadata.get("contact_id") if direct_id is None and subject.provider == "addresses" and subject.resource_type in { "contact", "address_contact", "address_email", }: direct_id = subject.resource_id if direct_id is not None: try: return [get_visible_contact(session, principal, str(direct_id))] except AddressBookError: try: resolution = resolve_contact_redirect( session, principal, str(direct_id), ) return [ get_visible_contact( session, principal, resolution.resolved_contact_id, ) ] except AddressBookError: return [] explicit_ref = subject.metadata.get("source_ref") source_refs = { str(explicit_ref).strip() if explicit_ref else "", f"{subject.provider}:{subject.resource_type}:{subject.resource_id}", } source_refs.discard("") rows = ( session.query(Contact) .options( selectinload(Contact.emails), selectinload(Contact.phones), selectinload(Contact.postal_addresses), selectinload(Contact.channel_rules), ) .filter( Contact.source_ref.in_(source_refs), or_(Contact.tenant_id == principal.tenant_id, Contact.tenant_id.is_(None)), ) .order_by(Contact.id.asc()) .limit(3) .all() ) visible: list[Contact] = [] visible_ids: set[str] = set() for row in rows: try: resolution = resolve_contact_redirect(session, principal, row.id) contact = get_visible_contact( session, principal, resolution.resolved_contact_id, ) if contact.id not in visible_ids: visible.append(contact) visible_ids.add(contact.id) except AddressBookError: continue return visible def _unresolved_contact_point_resolution( request: ContactPointResolutionRequest, *, code: str, message: str, status: str = "unresolved", provenance: dict[str, Any] | None = None, ) -> ContactPointResolution: return ContactPointResolution( contract_version=CONTACT_POINT_CONTRACT_VERSION, subject=request.subject, status=status, # type: ignore[arg-type] explanations=( DistributionExplanation( code=code, message=message, severity="warning", provider="addresses", source=request.subject, ), ), provenance={ "module": "addresses", "effective_at": request.effective_at.isoformat(), "purpose": request.purpose, **(provenance or {}), }, ) def _resolve_contact_points_for_contact( session: Any, principal: ApiPrincipal, *, request: ContactPointResolutionRequest, context: _SourceContactContext, ) -> ContactPointResolution: contact = context.contact facts = AddressesChannelFactsCapability().resolve_channel_facts( session, principal, request=RecipientChannelFactsRequest( tenant_id=request.tenant_id, source=DistributionSourceReference( provider="addresses", resource_type="contact", resource_id=contact.id, revision=contact.source_revision, label=contact.display_name, metadata={ "contact_id": contact.id, "source_ref": contact.source_ref, }, ), recipient_key=f"contact:{contact.id}", effective_at=request.effective_at, purpose=request.purpose, requested_channels=request.requested_channels, context=request.context, ), ) preference_revision = _rule_revision( contact, decisions={"preferred"}, ) consent_revision = _rule_revision( contact, decisions={ "allowed", "opted_in", "opted_out", "suppressed", "invalid", "returned", "temporarily_unavailable", }, ) candidates: list[ContactPointCandidate] = [] for fact in facts.candidates: if ( context.allowed_contact_point_ids is not None and ( fact.contact_point_id is None or fact.contact_point_id not in context.allowed_contact_point_ids ) ): continue details = _contact_point_details( contact, channel=fact.channel, contact_point_id=fact.contact_point_id, fallback_target=fact.target, fallback_target_key=fact.target_key, postal_format=request.postal_format, ) if details is None: continue address_purpose, primary, target, target_key, value, order_index = details preference_rank = fact.decision_provenance.get("preference_rank") candidates.append( ContactPointCandidate( channel=fact.channel, target=target, target_key=target_key, status=fact.status, contact_point_id=fact.contact_point_id, address_purpose=address_purpose, locale=fact.locale or request.locale, preferred=fact.preferred or primary, preference_rank=( int(preference_rank) if isinstance(preference_rank, int) else None ), reason_code=fact.reason_code, explanation=fact.explanation, source=fact.source, source_revision=facts.source_revision, preference_revision=preference_revision, consent_revision=consent_revision, value=value, provenance={ **dict(fact.decision_provenance), "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, "order_index": order_index, "address_list_entry_ids": context.address_list_entry_ids, }, ) ) selected, excluded = _select_contact_point_candidates(candidates, request) if selected: result_status = "usable" elif excluded: result_status = _resolution_status(excluded) else: result_status = "unresolved" explanations = list(facts.explanations) if not selected and not excluded: explanations.append( DistributionExplanation( code="addresses.contact_point.none", message="The contact has no contact point for the requested channels.", severity="warning", provider="addresses", source=request.subject, ) ) return ContactPointResolution( contract_version=CONTACT_POINT_CONTRACT_VERSION, subject=request.subject, status=result_status, # type: ignore[arg-type] contact_id=contact.id, display_name=contact.display_name, candidates=tuple(selected), excluded=tuple(excluded), explanations=tuple(explanations), source_revision=facts.source_revision, source_fingerprint=facts.source_fingerprint, provenance={ **dict(facts.provenance), "provider_subject": _json_value(asdict(request.subject)), "address_purpose": request.address_purpose, "fallback_rule": request.fallback_rule, "postal_format": request.postal_format, "preference_revision": preference_revision, "consent_revision": consent_revision, "address_list_entry_ids": context.address_list_entry_ids, }, ) def _contact_point_details( contact: Contact, *, channel: str, contact_point_id: str | None, fallback_target: str, fallback_target_key: str, postal_format: str, ) -> tuple[str | None, bool, str, str, dict[str, Any], int] | None: if channel == "email": point = next((item for item in contact.emails if item.id == contact_point_id), None) if point is None: return None target = point.email.strip() return ( point.label, point.is_primary, target, f"email:{target.casefold()}", {"email": target, "label": point.label}, point.order_index, ) if channel == "postal": point = next( (item for item in contact.postal_addresses if item.id == contact_point_id), None, ) if point is None: return None lines = [ contact.display_name, point.street, " ".join(part for part in (point.postal_code, point.locality) if part), point.region, ] if postal_format == "international": lines.append(point.country) normalized_lines = [str(item).strip() for item in lines if item and str(item).strip()] canonical = "|".join( str(item or "").strip().casefold() for item in ( point.street, point.postal_code, point.locality, point.region, point.country, ) ) return ( point.label, point.is_primary, "\n".join(normalized_lines), f"postal:{canonical}", { "label": point.label, "street": point.street, "postal_code": point.postal_code, "locality": point.locality, "region": point.region, "country": point.country, "formatted_lines": normalized_lines, "format": postal_format, }, point.order_index, ) if channel in {"internal_mail", "portal"}: purpose = (contact.provenance or {}).get(f"{channel}_purpose") return ( str(purpose) if purpose else None, True, fallback_target, fallback_target_key, {"reference": fallback_target}, 0, ) return None def _select_contact_point_candidates( candidates: list[ContactPointCandidate], request: ContactPointResolutionRequest, ) -> tuple[list[ContactPointCandidate], list[ContactPointCandidate]]: candidates = sorted(candidates, key=_contact_point_sort_key) accepted: list[ContactPointCandidate] = [] excluded: list[ContactPointCandidate] = [] requested_purpose = (request.address_purpose or "").strip().casefold() by_channel: dict[str, list[ContactPointCandidate]] = {} for item in candidates: by_channel.setdefault(item.channel, []).append(item) for channel_items in by_channel.values(): selected_ids: set[int] if not requested_purpose: selected_ids = {id(item) for item in channel_items} else: exact = [ item for item in channel_items if (item.address_purpose or "").strip().casefold() == requested_purpose ] if exact: selected_ids = {id(item) for item in exact} elif request.fallback_rule == "primary": primary = [item for item in channel_items if item.preferred] selected_ids = {id(item) for item in primary[:1]} elif request.fallback_rule == "any": selected_ids = {id(channel_items[0])} if channel_items else set() else: selected_ids = set() for item in channel_items: if id(item) in selected_ids: accepted.append(item) else: excluded.append( item if item.status not in {"usable", "stale"} else replace( item, status="unresolved", reason_code="addresses.address_purpose.not_selected", explanation=( "This contact point did not match the requested address purpose " "or its configured fallback." ), ) ) deduplicated: list[ContactPointCandidate] = [] seen: set[tuple[str, str]] = set() for item in accepted: key = (item.channel, item.target_key) if key in seen and item.status in {"usable", "stale"}: excluded.append( replace( item, status="duplicate", reason_code="addresses.contact_point.duplicate", explanation="An earlier contact point resolves to the same channel target.", ) ) continue seen.add(key) if item.status in {"usable", "stale"}: deduplicated.append(item) else: excluded.append(item) return deduplicated, sorted(excluded, key=_contact_point_sort_key) def _contact_point_sort_key(item: ContactPointCandidate) -> tuple[Any, ...]: order = item.provenance.get("order_index") return ( {"email": 0, "postal": 1, "internal_mail": 2, "portal": 3}.get( item.channel, 9, ), 0 if item.preferred else 1, item.preference_rank if item.preference_rank is not None else 10001, int(order) if isinstance(order, int) else 0, item.contact_point_id or "", item.target_key, ) def _resolution_status(excluded: list[ContactPointCandidate]) -> str: for status in ( "invalid", "suppressed", "ambiguous", "duplicate", "stale", "unresolved", ): if any(item.status == status for item in excluded): return status return "unresolved" def _rule_revision(contact: Contact, *, decisions: set[str]) -> str | None: stamps = [ _aware_datetime(item.updated_at) for item in contact.channel_rules if item.decision in decisions ] return max(stamps).isoformat() if stamps else None def _contact_point_source_contexts( session: Any, principal: ApiPrincipal, source_id: str, ) -> tuple[DistributionSourceReference, str, list[_SourceContactContext]]: book_prefix = "addresses:address_book:" list_prefix = "addresses:address_list:" if source_id.startswith(book_prefix): book = get_visible_address_book( session, principal, source_id.removeprefix(book_prefix), ) contexts = [ _SourceContactContext(contact=item) for item in _active_address_book_contacts(session, book.id) ] revision = _recipient_source_revision( book, _address_book_contact_updated_at(session, [book.id]).get(book.id), ) return ( DistributionSourceReference( provider="addresses", resource_type="address_book", resource_id=book.id, revision=revision, label=book.name, metadata={ "source_id": source_id, "scope_type": book.scope_type, "scope_id": book.scope_id, "source_kind": book.source_kind, "source_ref": book.source_ref, }, ), revision, contexts, ) if source_id.startswith(list_prefix): address_list = get_visible_address_list( session, principal, source_id.removeprefix(list_prefix), ) entries = list_address_list_entries(session, principal, address_list.id) grouped: dict[str, dict[str, Any]] = {} for entry in entries: item = grouped.setdefault( entry.contact_id, {"contact": entry.contact, "allowed": set(), "all": False, "entry_ids": []}, ) item["entry_ids"].append(entry.id) point_id = entry.contact_email_id or entry.contact_postal_address_id if point_id: item["allowed"].add(point_id) else: item["all"] = True contexts = [ _SourceContactContext( contact=item["contact"], allowed_contact_point_ids=( None if item["all"] else frozenset(item["allowed"]) ), address_list_entry_ids=tuple(item["entry_ids"]), ) for item in grouped.values() ] revision = _address_list_source_revision( address_list, _address_list_updated_at(session, [address_list.id]).get(address_list.id), ) return ( DistributionSourceReference( provider="addresses", resource_type="address_list", resource_id=address_list.id, revision=revision, label=address_list.name, metadata={ "source_id": source_id, "address_book_id": address_list.address_book_id, "scope_type": address_list.address_book.scope_type, "scope_id": address_list.address_book.scope_id, }, ), revision, contexts, ) raise ValueError(f"Unsupported Addresses contact-point source id: {source_id}") def _contact_point_resolution_request( request: ContactPointSourceRequest, contact: Contact, ) -> ContactPointResolutionRequest: return ContactPointResolutionRequest( tenant_id=request.tenant_id, subject=DistributionSourceReference( provider="addresses", resource_type="contact", resource_id=contact.id, revision=contact.source_revision, label=contact.display_name, metadata={ "contact_id": contact.id, "source_kind": contact.source_kind, "source_ref": contact.source_ref, }, ), effective_at=request.effective_at, purpose=request.purpose, requested_channels=request.requested_channels, address_purpose=request.address_purpose, fallback_rule=request.fallback_rule, locale=request.locale, postal_format=request.postal_format, context=request.context, ) def _source_fingerprint(source_id: str, revision: str) -> str: return hashlib.sha256(f"{source_id}\0{revision}".encode("utf-8")).hexdigest() def _json_value(value: Any) -> Any: if isinstance(value, datetime): return value.isoformat() if isinstance(value, dict): return {str(key): _json_value(item) for key, item in value.items()} if isinstance(value, (list, tuple, set, frozenset)): return [_json_value(item) for item in value] return value def _source_reference_from_payload(payload: dict[str, Any]) -> DistributionSourceReference: return DistributionSourceReference( provider=str(payload.get("provider") or "addresses"), resource_type=str(payload.get("resource_type") or "unknown"), resource_id=str(payload.get("resource_id") or ""), revision=str(payload["revision"]) if payload.get("revision") else None, fingerprint=str(payload["fingerprint"]) if payload.get("fingerprint") else None, label=str(payload["label"]) if payload.get("label") else None, metadata=dict(payload.get("metadata") or {}), ) def _candidate_from_payload(payload: dict[str, Any]) -> ContactPointCandidate: source_payload = payload.get("source") return ContactPointCandidate( channel=str(payload.get("channel") or "email"), # type: ignore[arg-type] target=str(payload.get("target") or ""), target_key=str(payload.get("target_key") or ""), status=str(payload.get("status") or "unresolved"), # type: ignore[arg-type] contact_point_id=( str(payload["contact_point_id"]) if payload.get("contact_point_id") else None ), address_purpose=( str(payload["address_purpose"]) if payload.get("address_purpose") else None ), locale=str(payload["locale"]) if payload.get("locale") else None, preferred=bool(payload.get("preferred")), preference_rank=( int(payload["preference_rank"]) if isinstance(payload.get("preference_rank"), int) else None ), reason_code=( str(payload["reason_code"]) if payload.get("reason_code") else None ), explanation=( str(payload["explanation"]) if payload.get("explanation") else None ), source=( _source_reference_from_payload(dict(source_payload)) if isinstance(source_payload, dict) else None ), source_revision=( str(payload["source_revision"]) if payload.get("source_revision") else None ), preference_revision=( str(payload["preference_revision"]) if payload.get("preference_revision") else None ), consent_revision=( str(payload["consent_revision"]) if payload.get("consent_revision") else None ), value=dict(payload.get("value") or {}), provenance=dict(payload.get("provenance") or {}), ) def _explanation_from_payload(payload: dict[str, Any]) -> DistributionExplanation: source_payload = payload.get("source") return DistributionExplanation( code=str(payload.get("code") or "addresses.snapshot"), message=str(payload.get("message") or ""), severity=str(payload.get("severity") or "warning"), # type: ignore[arg-type] provider=str(payload["provider"]) if payload.get("provider") else None, source=( _source_reference_from_payload(dict(source_payload)) if isinstance(source_payload, dict) else None ), provenance=dict(payload.get("provenance") or {}), ) def _resolution_from_payload(payload: dict[str, Any]) -> ContactPointResolution: return ContactPointResolution( contract_version=str( payload.get("contract_version") or CONTACT_POINT_CONTRACT_VERSION ), subject=_source_reference_from_payload(dict(payload.get("subject") or {})), status=str(payload.get("status") or "unresolved"), # type: ignore[arg-type] contact_id=str(payload["contact_id"]) if payload.get("contact_id") else None, display_name=( str(payload["display_name"]) if payload.get("display_name") else None ), candidates=tuple( _candidate_from_payload(dict(item)) for item in payload.get("candidates") or [] if isinstance(item, dict) ), excluded=tuple( _candidate_from_payload(dict(item)) for item in payload.get("excluded") or [] if isinstance(item, dict) ), explanations=tuple( _explanation_from_payload(dict(item)) for item in payload.get("explanations") or [] if isinstance(item, dict) ), source_revision=( str(payload["source_revision"]) if payload.get("source_revision") else None ), source_fingerprint=( str(payload["source_fingerprint"]) if payload.get("source_fingerprint") else None ), provenance=dict(payload.get("provenance") or {}), ) def _source_request_from_payload(payload: dict[str, Any]) -> ContactPointSourceRequest: effective_at = datetime.fromisoformat(str(payload["effective_at"])) return ContactPointSourceRequest( tenant_id=str(payload.get("tenant_id") or ""), source_id=str(payload.get("source_id") or ""), effective_at=effective_at, purpose=str(payload["purpose"]) if payload.get("purpose") else None, requested_channels=tuple(payload.get("requested_channels") or ()), # type: ignore[arg-type] address_purpose=( str(payload["address_purpose"]) if payload.get("address_purpose") else None ), fallback_rule=str(payload.get("fallback_rule") or "primary"), # type: ignore[arg-type] locale=str(payload["locale"]) if payload.get("locale") else None, postal_format=str(payload.get("postal_format") or "domestic"), # type: ignore[arg-type] max_items=int(payload.get("max_items") or 5000), context=dict(payload.get("context") or {}), ) def _contact_point_snapshot_ref(snapshot: ContactPointSnapshot) -> ContactPointSnapshotRef: provenance = dict(snapshot.provenance or {}) source_payload = provenance.get("source") source = _source_reference_from_payload( dict(source_payload) if isinstance(source_payload, dict) else {} ) return ContactPointSnapshotRef( id=snapshot.id, tenant_id=snapshot.tenant_id, contract_version=snapshot.contract_version, source=source, request=_source_request_from_payload(dict(snapshot.request_payload or {})), resolutions=tuple( _resolution_from_payload(dict(item)) for item in snapshot.resolution_payload or [] ), recipient_count=snapshot.recipient_count, excluded_count=snapshot.excluded_count, source_revision=snapshot.source_revision, source_fingerprint=snapshot.source_fingerprint, snapshot_hash=snapshot.snapshot_hash, generated_at=_aware_datetime(snapshot.generated_at), provenance=provenance, )