govoplan-addresses/src/govoplan_addresses/backend/service.py

from __future__ import annotations

import copy
from collections.abc import Iterable
from dataclasses import dataclass, field
import os
import re
import urllib.parse
from typing import Any

from sqlalchemy import and_, false, func, or_
from sqlalchemy.orm import Session

from govoplan_core.auth import ApiPrincipal
from govoplan_core.audit.logging import audit_event
from govoplan_core.core.change_sequence import record_change
from govoplan_core.db.base import utcnow
from govoplan_core.security.credential_envelopes import (
    CredentialAccessContext,
    CredentialEnvelopeError,
    ResolvedCredentialEnvelope,
    credential_envelope_summary,
    list_credential_envelopes,
    resolve_credential_envelope,
)
from govoplan_core.security.secrets import decrypt_secret, encrypt_secret
from govoplan_addresses.backend.carddav import (
    AddressCardDAVAddressBook,
    AddressCardDAVClient,
    AddressCardDAVError,
    AddressCardDAVObject,
    AddressCardDAVPreconditionFailed,
    AddressCardDAVReportResult,
    AddressCardDAVSyncUnsupported,
    ensure_collection_url,
)
from govoplan_addresses.backend.db.models import (
    AddressBook,
    AddressList,
    AddressListEntry,
    AddressSyncConflict,
    AddressSyncDiagnostic,
    AddressSyncSource,
    AddressSyncTombstone,
    Contact,
    ContactEmail,
    ContactPhone,
    ContactPostalAddress,
)
from govoplan_addresses.backend.schemas import (
    AddressBookCreateRequest,
    AddressBookUpdateRequest,
    AddressListCreateRequest,
    AddressListEntryCreateRequest,
    AddressListUpdateRequest,
    AddressCardDavDiscoveryRequest,
    AddressCardDavSourceCreateRequest,
    AddressSyncPlanStats,
    AddressSyncAttemptFinishRequest,
    AddressSyncConflictCreateRequest,
    AddressSyncConflictResolveRequest,
    AddressSyncDiagnosticCreateRequest,
    AddressSyncSourceCreateRequest,
    AddressSyncSourceUpdateRequest,
    AddressSyncTombstoneCreateRequest,
    ContactCreateRequest,
    ContactEmailPayload,
    ContactPhonePayload,
    ContactPostalAddressPayload,
    ContactUpdateRequest,
)
from govoplan_addresses.backend.vcard import ParsedVCard, ParsedVCardIssue, contacts_to_vcard, parse_vcards_with_issues


class AddressBookError(ValueError):
    pass


CORE_CREDENTIAL_ENVELOPE_PREFIX = "credential-envelope:"


def address_credential_context(
    *,
    tenant_id: str,
    source_id: str | None = None,
) -> CredentialAccessContext:
    return CredentialAccessContext(
        tenant_id=tenant_id,
        target_scope_type="tenant",
        target_scope_id=tenant_id,
        module_id="addresses",
        server_ref=f"addresses:{source_id}" if source_id else None,
    )


def available_address_credentials(
    session: Session,
    principal: ApiPrincipal,
    *,
    source_id: str | None = None,
) -> list[dict[str, Any]]:
    context = address_credential_context(
        tenant_id=principal.tenant_id,
        source_id=source_id,
    )
    return [
        credential_envelope_summary(row)
        for row in list_credential_envelopes(session, context=context)
    ]


@dataclass(frozen=True, slots=True)
class VCardImportResult:
    contacts: list[Contact]
    issues: list[ParsedVCardIssue]
    skipped: int


@dataclass(slots=True)
class AddressSyncPlanItem:
    action: str
    href: str | None = None
    remote_uid: str | None = None
    contact_id: str | None = None
    display_name: str | None = None
    etag: str | None = None
    message: str | None = None
    raw_vcard: str | None = None
    parsed_payload: ContactCreateRequest | None = None
    source_revision: str | None = None


@dataclass(slots=True)
class AddressSyncPlan:
    sync_source: AddressSyncSource
    stats: AddressSyncPlanStats
    items: list[AddressSyncPlanItem] = field(default_factory=list)


READ_ONLY_SYNC_DIRECTIONS = {"read_only", "import"}
OUTBOUND_SYNC_DIRECTIONS = {"export", "two_way"}
INBOUND_SYNC_DIRECTIONS = {"read_only", "import", "two_way"}
REMOTE_SYNC_ACTIONS = {"remote_create", "remote_update", "remote_delete"}
ADDRESS_MODULE_ID = "addresses"
ADDRESS_CONTACTS_COLLECTION = "addresses.contacts"
ADDRESS_CONTACT_RESOURCE = "address_contact"
CARDDAV_SECRET_ENV_PREFIX = "env:"  # noqa: S105  # nosec B105 - reference prefix, not a credential.


def _trim(value: str | None) -> str | None:
    if value is None:
        return None
    trimmed = value.strip()
    return trimmed or None


def _account_id(principal: ApiPrincipal) -> str:
    return principal.account_id


def _tenant_id_or_none(principal: ApiPrincipal) -> str | None:
    try:
        return principal.tenant_id
    except RuntimeError:
        return None


def _visible_book_predicate(principal: ApiPrincipal):
    tenant_id = _tenant_id_or_none(principal)
    account_id = _account_id(principal)
    predicates = [AddressBook.scope_type == "system"]
    if tenant_id:
        predicates.extend(
            [
                and_(AddressBook.tenant_id == tenant_id, AddressBook.scope_type == "tenant"),
                and_(AddressBook.tenant_id == tenant_id, AddressBook.scope_type == "user", AddressBook.scope_id == account_id),
            ]
        )
        group_ids = tuple(principal.group_ids)
        if group_ids:
            predicates.append(and_(AddressBook.tenant_id == tenant_id, AddressBook.scope_type == "group", AddressBook.scope_id.in_(group_ids)))
    return or_(*predicates)


def _visible_books_query(session: Session, principal: ApiPrincipal, *, include_deleted: bool = False):
    query = session.query(AddressBook).filter(_visible_book_predicate(principal))
    if not include_deleted:
        query = query.filter(AddressBook.deleted_at.is_(None))
    return query


def _require_mutable_book(book: AddressBook) -> None:
    if book.read_only:
        raise AddressBookError("Address book is read-only.")
    if book.deleted_at is not None:
        raise AddressBookError("Address book is deleted.")


def _require_mutable_address_list(address_list: AddressList) -> None:
    _require_mutable_book(address_list.address_book)
    if address_list.read_only:
        raise AddressBookError("Address list is read-only.")
    if address_list.deleted_at is not None:
        raise AddressBookError("Address list is deleted.")


def _scope_id_for_create(principal: ApiPrincipal, payload: AddressBookCreateRequest, *, allow_system: bool = False) -> tuple[str | None, str | None]:
    tenant_id = _tenant_id_or_none(principal)
    if payload.scope_type == "system":
        if not allow_system:
            raise AddressBookError("Creating system address books requires address book administration permission.")
        return None, None
    if not tenant_id:
        raise AddressBookError("A tenant context is required for tenant, group, and user address books.")
    if payload.scope_type == "tenant":
        return tenant_id, tenant_id
    if payload.scope_type == "user":
        return tenant_id, _account_id(principal)
    if payload.scope_type == "group":
        group_id = _trim(payload.group_id)
        if not group_id:
            raise AddressBookError("Group address books require a group id.")
        if group_id not in principal.group_ids and not principal.has("addresses:address_book:admin"):
            raise AddressBookError("The selected group is not visible to the current principal.")
        return tenant_id, group_id
    raise AddressBookError("Unsupported address book scope.")


def list_address_books(session: Session, principal: ApiPrincipal, *, include_deleted: bool = False) -> list[AddressBook]:
    return _visible_books_query(session, principal, include_deleted=include_deleted).order_by(AddressBook.scope_type.asc(), AddressBook.name.asc(), AddressBook.id.asc()).all()


def address_book_contact_counts(session: Session, book_ids: Iterable[str], *, include_deleted: bool = False) -> dict[str, int]:
    ids = tuple(book_ids)
    if not ids:
        return {}
    query = (
        session.query(Contact.address_book_id, func.count(Contact.id))
        .filter(Contact.address_book_id.in_(ids))
        .group_by(Contact.address_book_id)
    )
    if not include_deleted:
        query = query.filter(Contact.deleted_at.is_(None))
    rows = query.all()
    return {book_id: count for book_id, count in rows}


def address_list_entry_counts(session: Session, address_list_ids: Iterable[str]) -> dict[str, int]:
    ids = tuple(address_list_ids)
    if not ids:
        return {}
    rows = (
        session.query(AddressListEntry.address_list_id, func.count(AddressListEntry.id))
        .join(Contact, AddressListEntry.contact_id == Contact.id)
        .filter(AddressListEntry.address_list_id.in_(ids), Contact.deleted_at.is_(None))
        .group_by(AddressListEntry.address_list_id)
        .all()
    )
    return {address_list_id: count for address_list_id, count in rows}


def get_visible_address_book(session: Session, principal: ApiPrincipal, book_id: str, *, include_deleted: bool = False) -> AddressBook:
    book = _visible_books_query(session, principal, include_deleted=include_deleted).filter(AddressBook.id == book_id).one_or_none()
    if book is None:
        raise AddressBookError("Address book not found.")
    return book


def _visible_sync_source_query(session: Session, principal: ApiPrincipal):
    book_ids = [book.id for book in list_address_books(session, principal)]
    if not book_ids:
        return session.query(AddressSyncSource).filter(false())
    return session.query(AddressSyncSource).filter(AddressSyncSource.address_book_id.in_(book_ids))


def list_sync_sources(
    session: Session,
    principal: ApiPrincipal,
    *,
    address_book_id: str | None = None,
    include_disabled: bool = False,
) -> list[AddressSyncSource]:
    query = _visible_sync_source_query(session, principal)
    if address_book_id:
        get_visible_address_book(session, principal, address_book_id)
        query = query.filter(AddressSyncSource.address_book_id == address_book_id)
    if not include_disabled:
        query = query.filter(AddressSyncSource.enabled.is_(True))
    return query.order_by(AddressSyncSource.display_name.asc(), AddressSyncSource.id.asc()).all()


def get_visible_sync_source(session: Session, principal: ApiPrincipal, sync_source_id: str) -> AddressSyncSource:
    sync_source = _visible_sync_source_query(session, principal).filter(AddressSyncSource.id == sync_source_id).one_or_none()
    if sync_source is None:
        raise AddressBookError("Address sync source not found.")
    return sync_source


def create_sync_source(
    session: Session,
    principal: ApiPrincipal,
    address_book_id: str,
    payload: AddressSyncSourceCreateRequest,
    *,
    trusted_connector_metadata: bool = False,
) -> AddressSyncSource:
    book = get_visible_address_book(session, principal, address_book_id)
    if book.deleted_at is not None:
        raise AddressBookError("Address book is deleted.")
    connector_type = _trim(payload.connector_type)
    display_name = _trim(payload.display_name)
    if not connector_type:
        raise AddressBookError("Sync connector type is required.")
    if not display_name:
        raise AddressBookError("Sync source display name is required.")
    if connector_type.casefold() == "carddav" and not trusted_connector_metadata:
        _assert_api_carddav_metadata_safe(payload.metadata)
    read_only = _read_only_from_sync_direction(payload.sync_direction, payload.read_only)
    sync_source = AddressSyncSource(
        tenant_id=book.tenant_id,
        address_book_id=book.id,
        connector_type=connector_type,
        display_name=display_name,
        external_account_ref=_trim(payload.external_account_ref),
        external_address_book_ref=_trim(payload.external_address_book_ref),
        sync_direction=payload.sync_direction,
        read_only=read_only,
        enabled=payload.enabled,
        status="idle" if payload.enabled else "disabled",
        sync_token=payload.sync_token,
        etag=_trim(payload.etag),
        remote_revision=_trim(payload.remote_revision),
        created_by_account_id=_account_id(principal),
        updated_by_account_id=_account_id(principal),
        metadata_=payload.metadata or {},
    )
    session.add(sync_source)
    session.flush()
    _apply_sync_source_to_book(book, sync_source)
    return sync_source


def update_sync_source(
    session: Session,
    principal: ApiPrincipal,
    sync_source_id: str,
    payload: AddressSyncSourceUpdateRequest,
) -> AddressSyncSource:
    sync_source = get_visible_sync_source(session, principal, sync_source_id)
    if "display_name" in payload.model_fields_set:
        display_name = _trim(payload.display_name)
        if not display_name:
            raise AddressBookError("Sync source display name is required.")
        sync_source.display_name = display_name
    if "external_account_ref" in payload.model_fields_set:
        sync_source.external_account_ref = _trim(payload.external_account_ref)
    if "external_address_book_ref" in payload.model_fields_set:
        sync_source.external_address_book_ref = _trim(payload.external_address_book_ref)
    if "sync_direction" in payload.model_fields_set and payload.sync_direction is not None:
        sync_source.sync_direction = payload.sync_direction
        sync_source.read_only = _read_only_from_sync_direction(payload.sync_direction, payload.read_only)
    elif "read_only" in payload.model_fields_set and payload.read_only is not None:
        sync_source.read_only = _read_only_from_sync_direction(sync_source.sync_direction, payload.read_only)
    if "enabled" in payload.model_fields_set and payload.enabled is not None:
        sync_source.enabled = payload.enabled
        if not payload.enabled:
            sync_source.status = "disabled"
        elif sync_source.status == "disabled":
            sync_source.status = "idle"
    if "sync_token" in payload.model_fields_set:
        sync_source.sync_token = payload.sync_token
    if "etag" in payload.model_fields_set:
        sync_source.etag = _trim(payload.etag)
    if "remote_revision" in payload.model_fields_set:
        sync_source.remote_revision = _trim(payload.remote_revision)
    if "metadata" in payload.model_fields_set:
        metadata = payload.metadata or {}
        if sync_source.connector_type.casefold() == "carddav":
            _assert_api_carddav_metadata_safe(metadata)
            metadata = _merge_server_owned_carddav_metadata(sync_source.metadata_, metadata)
        sync_source.metadata_ = metadata
    sync_source.updated_by_account_id = _account_id(principal)
    _apply_sync_source_to_book(sync_source.address_book, sync_source)
    return sync_source


def delete_sync_source(session: Session, principal: ApiPrincipal, sync_source_id: str) -> None:
    sync_source = get_visible_sync_source(session, principal, sync_source_id)
    book = sync_source.address_book
    if book.source_ref == sync_source.id:
        book.source_kind = "local"
        book.source_ref = None
        book.read_only = False
        book.sync_status = None
        book.sync_error = None
        book.updated_by_account_id = _account_id(principal)
    _audit_sync_credential_deletion(session, principal, sync_source)
    session.delete(sync_source)


def _audit_sync_credential_deletion(
    session: Session,
    principal: ApiPrincipal,
    sync_source: AddressSyncSource,
) -> None:
    """Audit removal of source-owned credential material in the DB transaction.

    CardDAV credentials are encrypted inside the sync-source row. Deleting that
    row therefore deletes the credential atomically with the connector. Legacy
    credential references are detached from the source, but are never resolved
    or sent to an external provider because ownership cannot be proven.
    """

    tenant_id = _tenant_id_or_none(principal)
    user = getattr(principal, "user", None)
    _record_sync_credential_deletion_audit(
        session,
        sync_source=sync_source,
        tenant_id=tenant_id,
        user_id=getattr(user, "id", None),
        api_key_id=getattr(principal, "api_key_id", None),
        deletion_reason="sync_source_deleted",
    )


def _record_sync_credential_deletion_audit(
    session: Session,
    *,
    sync_source: AddressSyncSource,
    tenant_id: str | None,
    user_id: str | None,
    api_key_id: str | None,
    deletion_reason: str,
) -> bool:
    auth = _carddav_auth_metadata(sync_source.metadata_)
    if auth.get("secret_encrypted"):
        storage_backend = "encrypted_database"
    elif auth.get("credential_ref"):
        storage_backend = "legacy_reference"
    else:
        return False
    audit_event(
        session,
        tenant_id=tenant_id,
        user_id=user_id,
        api_key_id=api_key_id,
        action="addresses.sync_credential_deleted",
        scope="tenant" if tenant_id is not None else "system",
        object_type="address_sync_credential",
        object_id=sync_source.id,
        details={
            "sync_source_id": sync_source.id,
            "connector_type": sync_source.connector_type,
            "storage_backend": storage_backend,
            "deletion_reason": deletion_reason,
        },
    )
    return True


def audit_address_credentials_for_retirement(session: Session) -> int:
    """Audit credentials that the ensuing destructive table drop deletes."""

    sources = session.query(AddressSyncSource).order_by(AddressSyncSource.id.asc()).all()
    audited = 0
    for source in sources:
        if _record_sync_credential_deletion_audit(
            session,
            sync_source=source,
            tenant_id=source.tenant_id,
            user_id=None,
            api_key_id=None,
            deletion_reason="module_data_retired",
        ):
            audited += 1
    session.flush()
    return audited


def start_sync_attempt(session: Session, principal: ApiPrincipal, sync_source_id: str) -> AddressSyncSource:
    sync_source = get_visible_sync_source(session, principal, sync_source_id)
    if not sync_source.enabled:
        raise AddressBookError("Address sync source is disabled.")
    sync_source.status = "running"
    sync_source.last_attempted_at = utcnow()
    sync_source.last_error = None
    sync_source.updated_by_account_id = _account_id(principal)
    sync_source.address_book.sync_status = "running"
    sync_source.address_book.sync_error = None
    sync_source.address_book.updated_by_account_id = _account_id(principal)
    return sync_source


def finish_sync_attempt(
    session: Session,
    principal: ApiPrincipal,
    sync_source_id: str,
    payload: AddressSyncAttemptFinishRequest,
) -> AddressSyncSource:
    sync_source = get_visible_sync_source(session, principal, sync_source_id)
    sync_source.status = payload.status
    sync_source.last_attempted_at = sync_source.last_attempted_at or utcnow()
    sync_source.last_error = _trim(payload.error)
    sync_source.last_diagnostic = payload.diagnostic or None
    if payload.status == "succeeded":
        sync_source.last_success_at = utcnow()
        sync_source.last_error = None
    if "sync_token" in payload.model_fields_set:
        sync_source.sync_token = payload.sync_token
    if "etag" in payload.model_fields_set:
        sync_source.etag = _trim(payload.etag)
    if "remote_revision" in payload.model_fields_set:
        sync_source.remote_revision = _trim(payload.remote_revision)
    sync_source.updated_by_account_id = _account_id(principal)
    sync_source.address_book.sync_status = payload.status
    sync_source.address_book.sync_error = sync_source.last_error
    sync_source.address_book.updated_by_account_id = _account_id(principal)
    return sync_source


def list_sync_diagnostics(
    session: Session,
    principal: ApiPrincipal,
    sync_source_id: str,
    *,
    limit: int = 100,
) -> list[AddressSyncDiagnostic]:
    sync_source = get_visible_sync_source(session, principal, sync_source_id)
    return (
        session.query(AddressSyncDiagnostic)
        .filter(AddressSyncDiagnostic.sync_source_id == sync_source.id)
        .order_by(AddressSyncDiagnostic.created_at.desc(), AddressSyncDiagnostic.id.desc())
        .limit(max(1, min(limit, 500)))
        .all()
    )


def record_sync_diagnostic(
    session: Session,
    principal: ApiPrincipal,
    sync_source_id: str,
    payload: AddressSyncDiagnosticCreateRequest,
) -> AddressSyncDiagnostic:
    sync_source = get_visible_sync_source(session, principal, sync_source_id)
    diagnostic = AddressSyncDiagnostic(
        tenant_id=sync_source.tenant_id,
        sync_source_id=sync_source.id,
        severity=payload.severity,
        code=_trim(payload.code) or "sync",
        message=_trim(payload.message) or "Sync diagnostic",
        details=payload.details or {},
    )
    sync_source.last_diagnostic = {
        "severity": diagnostic.severity,
        "code": diagnostic.code,
        "message": diagnostic.message,
    }
    if diagnostic.severity == "error":
        sync_source.last_error = diagnostic.message
        sync_source.status = "failed"
        sync_source.address_book.sync_status = "failed"
        sync_source.address_book.sync_error = diagnostic.message
    sync_source.updated_by_account_id = _account_id(principal)
    session.add(diagnostic)
    return diagnostic


def list_sync_tombstones(
    session: Session,
    principal: ApiPrincipal,
    sync_source_id: str,
    *,
    limit: int = 200,
) -> list[AddressSyncTombstone]:
    sync_source = get_visible_sync_source(session, principal, sync_source_id)
    return (
        session.query(AddressSyncTombstone)
        .filter(AddressSyncTombstone.sync_source_id == sync_source.id)
        .order_by(AddressSyncTombstone.created_at.desc(), AddressSyncTombstone.id.desc())
        .limit(max(1, min(limit, 1000)))
        .all()
    )


def record_sync_tombstone(
    session: Session,
    principal: ApiPrincipal,
    sync_source_id: str,
    payload: AddressSyncTombstoneCreateRequest,
) -> AddressSyncTombstone:
    sync_source = get_visible_sync_source(session, principal, sync_source_id)
    contact = _sync_contact_or_none(session, principal, sync_source, payload.contact_id)
    if not _trim(payload.remote_uid) and not _trim(payload.resource_href) and contact is None:
        raise AddressBookError("Sync tombstone requires a contact id, remote uid, or resource href.")
    tombstone = AddressSyncTombstone(
        tenant_id=sync_source.tenant_id,
        sync_source_id=sync_source.id,
        address_book_id=sync_source.address_book_id,
        contact_id=contact.id if contact is not None else None,
        remote_uid=_trim(payload.remote_uid),
        resource_href=_trim(payload.resource_href),
        local_deleted_at=payload.local_deleted_at,
        remote_deleted_at=payload.remote_deleted_at,
        synced_at=payload.synced_at,
        metadata_=payload.metadata or {},
    )
    session.add(tombstone)
    return tombstone


def list_sync_conflicts(
    session: Session,
    principal: ApiPrincipal,
    sync_source_id: str,
    *,
    status_filter: str | None = "open",
    limit: int = 200,
) -> list[AddressSyncConflict]:
    sync_source = get_visible_sync_source(session, principal, sync_source_id)
    query = session.query(AddressSyncConflict).filter(AddressSyncConflict.sync_source_id == sync_source.id)
    if status_filter:
        query = query.filter(AddressSyncConflict.status == status_filter)
    return query.order_by(AddressSyncConflict.created_at.desc(), AddressSyncConflict.id.desc()).limit(max(1, min(limit, 1000))).all()


def record_sync_conflict(
    session: Session,
    principal: ApiPrincipal,
    sync_source_id: str,
    payload: AddressSyncConflictCreateRequest,
) -> AddressSyncConflict:
    sync_source = get_visible_sync_source(session, principal, sync_source_id)
    contact = _sync_contact_or_none(session, principal, sync_source, payload.contact_id)
    field_path = _trim(payload.field_path)
    if not field_path:
        raise AddressBookError("Sync conflict field path is required.")
    conflict = AddressSyncConflict(
        tenant_id=sync_source.tenant_id,
        sync_source_id=sync_source.id,
        address_book_id=sync_source.address_book_id,
        contact_id=contact.id if contact is not None else None,
        remote_uid=_trim(payload.remote_uid),
        resource_href=_trim(payload.resource_href),
        field_path=field_path,
        local_value=payload.local_value,
        remote_value=payload.remote_value,
        local_updated_at=payload.local_updated_at,
        remote_updated_at=payload.remote_updated_at,
        status="open",
        metadata_=payload.metadata or {},
    )
    sync_source.status = "conflict"
    sync_source.address_book.sync_status = "conflict"
    sync_source.updated_by_account_id = _account_id(principal)
    session.add(conflict)
    return conflict


def resolve_sync_conflict(
    session: Session,
    principal: ApiPrincipal,
    conflict_id: str,
    payload: AddressSyncConflictResolveRequest,
) -> AddressSyncConflict:
    conflict = get_visible_sync_conflict(session, principal, conflict_id)
    if payload.status == "resolved" and payload.resolution == "use_remote":
        _apply_remote_sync_conflict(session, principal, conflict)
    elif payload.status == "resolved" and payload.resolution == "merge":
        if payload.merged_payload is None:
            raise AddressBookError("Merge resolution requires a merged contact payload.")
        _apply_payload_sync_conflict(session, principal, conflict, payload.merged_payload)
    conflict.status = payload.status
    conflict.resolution = payload.resolution
    conflict.resolved_at = utcnow()
    conflict.resolved_by_account_id = _account_id(principal)
    if not list_sync_conflicts(session, principal, conflict.sync_source_id, status_filter="open", limit=1):
        conflict.sync_source.status = "idle"
        conflict.address_book.sync_status = "idle"
    return conflict


def _apply_remote_sync_conflict(session: Session, principal: ApiPrincipal, conflict: AddressSyncConflict) -> None:
    metadata = dict(conflict.metadata_ or {})
    remote_value = dict(conflict.remote_value or {})
    raw_payload = metadata.get("remote_payload") or remote_value.get("payload")
    if not isinstance(raw_payload, dict):
        raise AddressBookError("This conflict does not include a stored remote payload that can be applied automatically.")
    payload = ContactCreateRequest.model_validate(raw_payload)
    _apply_payload_sync_conflict(session, principal, conflict, payload)


def _apply_payload_sync_conflict(session: Session, principal: ApiPrincipal, conflict: AddressSyncConflict, payload: ContactCreateRequest) -> None:
    metadata = dict(conflict.metadata_ or {})
    remote_value = dict(conflict.remote_value or {})
    item = AddressSyncPlanItem(
        action="update",
        href=conflict.resource_href,
        remote_uid=conflict.remote_uid,
        contact_id=conflict.contact_id,
        display_name=payload.display_name,
        etag=str(remote_value.get("etag") or "") or None,
        raw_vcard=metadata.get("raw_vcard") if isinstance(metadata.get("raw_vcard"), str) else None,
        parsed_payload=payload,
        source_revision=str(metadata.get("source_revision") or remote_value.get("etag") or "") or None,
    )
    _upsert_carddav_contact(session, principal, conflict.sync_source, item)


def get_visible_sync_conflict(session: Session, principal: ApiPrincipal, conflict_id: str) -> AddressSyncConflict:
    sync_source_ids = [source.id for source in list_sync_sources(session, principal, include_disabled=True)]
    if not sync_source_ids:
        raise AddressBookError("Address sync conflict not found.")
    conflict = session.query(AddressSyncConflict).filter(AddressSyncConflict.id == conflict_id, AddressSyncConflict.sync_source_id.in_(sync_source_ids)).one_or_none()
    if conflict is None:
        raise AddressBookError("Address sync conflict not found.")
    return conflict


def discover_carddav_address_books(
    session: Session,
    principal: ApiPrincipal,
    payload: AddressCardDavDiscoveryRequest,
) -> list[AddressCardDAVAddressBook]:
    _assert_no_caller_carddav_credential_ref(payload.credential_ref)
    source = get_visible_sync_source(session, principal, payload.source_id) if payload.source_id else None
    client = _carddav_client_from_payload(session, principal, payload, source=source)
    return client.discover_addressbooks()


def create_carddav_sync_source(
    session: Session,
    principal: ApiPrincipal,
    address_book_id: str,
    payload: AddressCardDavSourceCreateRequest,
) -> AddressSyncSource:
    _assert_no_caller_carddav_credential_ref(payload.credential_ref)
    if payload.credential_ref and (payload.password is not None or payload.bearer_token is not None):
        raise AddressBookError("Select a reusable credential or enter a new secret, not both.")
    collection_url = ensure_collection_url(payload.collection_url)
    display_name = _trim(payload.display_name) or "CardDAV address book"
    metadata = _carddav_metadata(
        auth_type=payload.auth_type,
        username=payload.username,
        password=_secret_value(payload.password),
        bearer_token=_secret_value(payload.bearer_token),
        credential_ref=payload.credential_ref,
        collection_url=collection_url,
    )
    source = create_sync_source(
        session,
        principal,
        address_book_id,
        AddressSyncSourceCreateRequest(
            connector_type="carddav",
            display_name=display_name,
            external_address_book_ref=collection_url,
            sync_direction=payload.sync_direction,
            read_only=payload.read_only,
            sync_token=payload.sync_token,
            etag=payload.etag,
            remote_revision=payload.remote_revision,
            metadata=metadata,
        ),
        trusted_connector_metadata=True,
    )
    reusable = _resolve_core_address_credential(
        session,
        tenant_id=principal.tenant_id,
        source_id=source.id,
        credential_ref=payload.credential_ref,
    )
    if reusable is not None:
        source_metadata = dict(source.metadata_ or {})
        auth = dict(source_metadata.get("carddav") or {})
        auth["username"] = _credential_username(reusable) or auth.get("username")
        source_metadata["carddav"] = auth
        source.metadata_ = source_metadata
    return source


def preview_sync_source(
    session: Session,
    principal: ApiPrincipal,
    sync_source_id: str,
    *,
    force_full: bool = False,
    password: str | None = None,
    bearer_token: str | None = None,
    client: AddressCardDAVClient | None = None,
) -> AddressSyncPlan:
    sync_source = get_visible_sync_source(session, principal, sync_source_id)
    if sync_source.connector_type != "carddav":
        raise AddressBookError(f"Preview is not implemented for {sync_source.connector_type} sync sources.")
    return _build_carddav_sync_plan(
        session,
        principal,
        sync_source,
        force_full=force_full,
        password=password,
        bearer_token=bearer_token,
        client=client,
    )


def run_sync_source(
    session: Session,
    principal: ApiPrincipal,
    sync_source_id: str,
    *,
    force_full: bool = False,
    password: str | None = None,
    bearer_token: str | None = None,
    client: AddressCardDAVClient | None = None,
) -> AddressSyncPlan:
    sync_source = start_sync_attempt(session, principal, sync_source_id)
    write_client = client
    if sync_source.connector_type == "carddav" and write_client is None:
        write_client = _carddav_client_for_source(session, sync_source, password=password, bearer_token=bearer_token)
    try:
        plan = _build_carddav_sync_plan(
            session,
            principal,
            sync_source,
            force_full=force_full,
            password=password,
            bearer_token=bearer_token,
            client=write_client,
        )
        _apply_address_sync_plan(session, principal, plan, client=write_client)
        status = "conflict" if plan.stats.conflicts else "succeeded"
        if plan.stats.errors:
            status = "failed"
        finish_sync_attempt(
            session,
            principal,
            sync_source.id,
            AddressSyncAttemptFinishRequest(
                status=status,
                sync_token=plan.stats.sync_token,
                etag=plan.stats.etag,
                remote_revision=plan.stats.remote_revision,
                error=f"{plan.stats.errors} sync item errors" if plan.stats.errors else None,
                diagnostic=_sync_plan_diagnostic(plan),
            ),
        )
        return plan
    except Exception as exc:
        finish_sync_attempt(
            session,
            principal,
            sync_source.id,
            AddressSyncAttemptFinishRequest(status="failed", error=str(exc), diagnostic={"error": str(exc)}),
        )
        raise


def _build_carddav_sync_plan(
    session: Session,
    principal: ApiPrincipal,
    sync_source: AddressSyncSource,
    *,
    force_full: bool,
    password: str | None,
    bearer_token: str | None,
    client: AddressCardDAVClient | None,
) -> AddressSyncPlan:
    client = client or _carddav_client_for_source(session, sync_source, password=password, bearer_token=bearer_token)
    stats = AddressSyncPlanStats()
    plan = AddressSyncPlan(sync_source=sync_source, stats=stats)
    collection_props = AddressCardDAVReportResult()
    try:
        try:
            collection_props = client.propfind_collection()
        except AddressCardDAVError as exc:
            plan.items.append(AddressSyncPlanItem(action="unchanged", message=f"Collection PROPFIND failed; continuing with REPORT: {exc}"))

        if sync_source.sync_token and not force_full:
            try:
                report = client.sync_collection(sync_source.sync_token)
                stats.used_sync_token = True
            except AddressCardDAVSyncUnsupported as exc:
                plan.items.append(AddressSyncPlanItem(action="unchanged", message=f"Sync token REPORT failed; falling back to full sync: {exc}"))
                report = client.list_objects()
                stats.full_sync = True
        else:
            report = client.list_objects()
            stats.full_sync = True

        stats.sync_token = report.sync_token or collection_props.sync_token or sync_source.sync_token
        stats.etag = report.ctag or collection_props.ctag or sync_source.etag
        stats.remote_revision = report.ctag or collection_props.ctag or sync_source.remote_revision
        _plan_carddav_report(session, sync_source=sync_source, client=client, report=report, plan=plan)
    except AddressCardDAVError as exc:
        plan.items.append(AddressSyncPlanItem(action="error", message=str(exc)))
        stats.errors += 1
    return plan


def _plan_carddav_report(
    session: Session,
    *,
    sync_source: AddressSyncSource,
    client: AddressCardDAVClient,
    report: AddressCardDAVReportResult,
    plan: AddressSyncPlan,
) -> None:
    local_by_href = _carddav_contacts_by_href(session, sync_source)
    seen_hrefs: set[str] = set()
    reads_remote = _sync_source_reads_remote(sync_source)
    for item in report.objects:
        href = item.href
        seen_hrefs.add(href)
        _plan_carddav_report_object(
            sync_source=sync_source,
            client=client,
            item=item,
            local=local_by_href.get(href),
            reads_remote=reads_remote,
            plan=plan,
        )

    if plan.stats.full_sync:
        _plan_carddav_full_sync_absences(
            sync_source=sync_source,
            local_by_href=local_by_href,
            seen_hrefs=seen_hrefs,
            reads_remote=reads_remote,
            plan=plan,
        )

    _plan_carddav_outbound_local_changes(session, sync_source=sync_source, plan=plan)


def _plan_carddav_report_object(
    *,
    sync_source: AddressSyncSource,
    client: AddressCardDAVClient,
    item: AddressCardDAVObject,
    local: Contact | None,
    reads_remote: bool,
    plan: AddressSyncPlan,
) -> None:
    if item.deleted:
        _plan_carddav_remote_deletion(
            sync_source=sync_source,
            item=item,
            local=local,
            reads_remote=reads_remote,
            plan=plan,
        )
        return
    loaded = _load_carddav_report_vcard(client=client, item=item, plan=plan)
    if loaded is None:
        return
    raw_vcard, parsed = loaded
    _plan_carddav_live_object(
        sync_source=sync_source,
        item=item,
        local=local,
        reads_remote=reads_remote,
        raw_vcard=raw_vcard,
        parsed=parsed,
        plan=plan,
    )


def _plan_carddav_remote_deletion(
    *,
    sync_source: AddressSyncSource,
    item: AddressCardDAVObject,
    local: Contact | None,
    reads_remote: bool,
    plan: AddressSyncPlan,
) -> None:
    if local is None or local.deleted_at is not None:
        _add_sync_plan_item(
            plan,
            AddressSyncPlanItem(
                action="unchanged",
                href=item.href,
                etag=item.etag,
                message="Remote delete already reflected locally.",
            ),
        )
        return
    if _sync_source_writes_remote(sync_source) and _local_contact_changed_after_last_sync(local, sync_source):
        _add_sync_plan_item(
            plan,
            AddressSyncPlanItem(
                action="conflict",
                href=item.href,
                contact_id=local.id,
                display_name=local.display_name,
                etag=item.etag,
                message="Remote object was deleted while the local contact changed since the last successful sync.",
            ),
        )
    elif reads_remote:
        _add_sync_plan_item(
            plan,
            AddressSyncPlanItem(
                action="delete",
                href=item.href,
                contact_id=local.id,
                display_name=local.display_name,
                etag=item.etag,
            ),
        )


def _load_carddav_report_vcard(
    *,
    client: AddressCardDAVClient,
    item: AddressCardDAVObject,
    plan: AddressSyncPlan,
) -> tuple[str, ParsedVCard] | None:
    raw_vcard = item.address_data
    if not raw_vcard:
        try:
            raw_vcard = client.fetch_object(item.href)
        except AddressCardDAVError as exc:
            _add_sync_plan_item(
                plan,
                AddressSyncPlanItem(action="error", href=item.href, etag=item.etag, message=str(exc)),
            )
            return None
    parsed, error_message = _parse_single_remote_vcard(raw_vcard)
    if parsed is None:
        _add_sync_plan_item(
            plan,
            AddressSyncPlanItem(
                action="error",
                href=item.href,
                etag=item.etag,
                message=error_message or "Remote vCard could not be parsed.",
            ),
        )
        return None
    return raw_vcard, parsed


def _plan_carddav_live_object(
    *,
    sync_source: AddressSyncSource,
    item: AddressCardDAVObject,
    local: Contact | None,
    reads_remote: bool,
    raw_vcard: str,
    parsed: ParsedVCard,
    plan: AddressSyncPlan,
) -> None:
    if local is None:
        _plan_new_carddav_remote_object(
            item=item,
            reads_remote=reads_remote,
            raw_vcard=raw_vcard,
            parsed=parsed,
            plan=plan,
        )
        return
    remote_revision = item.etag or parsed.source_revision
    if _carddav_local_delete_needs_push(local, sync_source):
        _add_sync_plan_item(
            plan,
            AddressSyncPlanItem(
                action="remote_delete",
                href=item.href,
                remote_uid=parsed.source_ref,
                contact_id=local.id,
                display_name=local.display_name,
                etag=item.etag or local.source_revision,
                message="Local delete will be pushed to CardDAV.",
            ),
        )
        return
    if local.deleted_at is None and local.source_revision == remote_revision:
        _plan_matching_carddav_revision(
            sync_source=sync_source,
            item=item,
            local=local,
            parsed=parsed,
            plan=plan,
        )
        return
    if _local_contact_changed_after_last_sync(local, sync_source):
        _add_sync_plan_item(
            plan,
            AddressSyncPlanItem(
                action="conflict",
                href=item.href,
                remote_uid=parsed.source_ref,
                contact_id=local.id,
                display_name=local.display_name,
                etag=item.etag,
                raw_vcard=raw_vcard,
                parsed_payload=parsed.payload,
                source_revision=item.etag or parsed.source_revision,
                message="Local contact changed since the last successful sync.",
            ),
        )
        return
    _plan_carddav_remote_update(
        item=item,
        local=local,
        reads_remote=reads_remote,
        raw_vcard=raw_vcard,
        parsed=parsed,
        remote_revision=remote_revision,
        plan=plan,
    )


def _plan_new_carddav_remote_object(
    *,
    item: AddressCardDAVObject,
    reads_remote: bool,
    raw_vcard: str,
    parsed: ParsedVCard,
    plan: AddressSyncPlan,
) -> None:
    if reads_remote:
        _add_sync_plan_item(
            plan,
            AddressSyncPlanItem(
                action="create",
                href=item.href,
                remote_uid=parsed.source_ref,
                display_name=parsed.payload.display_name,
                etag=item.etag,
                raw_vcard=raw_vcard,
                parsed_payload=parsed.payload,
                source_revision=item.etag or parsed.source_revision,
            ),
        )
    else:
        _add_sync_plan_item(
            plan,
            AddressSyncPlanItem(
                action="unchanged",
                href=item.href,
                etag=item.etag,
                message="Remote object ignored by export-only sync source.",
            ),
        )


def _carddav_local_delete_needs_push(local: Contact, sync_source: AddressSyncSource) -> bool:
    return bool(
        local.deleted_at is not None
        and _sync_source_writes_remote(sync_source)
        and _local_contact_changed_after_last_sync(local, sync_source)
    )


def _plan_matching_carddav_revision(
    *,
    sync_source: AddressSyncSource,
    item: AddressCardDAVObject,
    local: Contact,
    parsed: ParsedVCard,
    plan: AddressSyncPlan,
) -> None:
    if _sync_source_writes_remote(sync_source) and _local_contact_changed_after_last_sync(local, sync_source):
        _add_sync_plan_item(
            plan,
            AddressSyncPlanItem(
                action="remote_update",
                href=item.href,
                remote_uid=parsed.source_ref,
                contact_id=local.id,
                display_name=local.display_name,
                etag=item.etag or local.source_revision,
                raw_vcard=_carddav_contact_vcard(local, href=item.href),
                source_revision=item.etag or local.source_revision,
                message="Local update will be pushed to CardDAV.",
            ),
        )
        return
    _add_sync_plan_item(
        plan,
        AddressSyncPlanItem(
            action="unchanged",
            href=item.href,
            remote_uid=parsed.source_ref,
            contact_id=local.id,
            display_name=local.display_name,
            etag=item.etag,
        ),
    )
def _plan_carddav_remote_update(
    *,
    item: AddressCardDAVObject,
    local: Contact,
    reads_remote: bool,
    raw_vcard: str,
    parsed: ParsedVCard,
    remote_revision: str | None,
    plan: AddressSyncPlan,
) -> None:
    if reads_remote:
        _add_sync_plan_item(
            plan,
            AddressSyncPlanItem(
                action="update",
                href=item.href,
                remote_uid=parsed.source_ref,
                contact_id=local.id,
                display_name=parsed.payload.display_name or local.display_name,
                etag=item.etag,
                raw_vcard=raw_vcard,
                parsed_payload=parsed.payload,
                source_revision=remote_revision,
            ),
        )
    else:
        _add_sync_plan_item(
            plan,
            AddressSyncPlanItem(
                action="unchanged",
                href=item.href,
                contact_id=local.id,
                display_name=local.display_name,
                etag=item.etag,
                message="Remote update ignored by export-only sync source.",
            ),
        )


def _plan_carddav_full_sync_absences(
    *,
    sync_source: AddressSyncSource,
    local_by_href: dict[str, Contact],
    seen_hrefs: set[str],
    reads_remote: bool,
    plan: AddressSyncPlan,
) -> None:
    for href, contact in local_by_href.items():
        if href in seen_hrefs or contact.deleted_at is not None:
            continue
        if _sync_source_writes_remote(sync_source) and _local_contact_changed_after_last_sync(contact, sync_source):
            _add_sync_plan_item(
                plan,
                AddressSyncPlanItem(
                    action="conflict",
                    href=href,
                    contact_id=contact.id,
                    display_name=contact.display_name,
                    etag=contact.source_revision,
                    message="Remote object is absent from full sync, but the local contact changed since the last successful sync.",
                ),
            )
        elif reads_remote:
            _add_sync_plan_item(
                plan,
                AddressSyncPlanItem(
                    action="delete",
                    href=href,
                    contact_id=contact.id,
                    display_name=contact.display_name,
                    message="Remote object is absent from full sync.",
                ),
            )


def _plan_carddav_outbound_local_changes(
    session: Session,
    *,
    sync_source: AddressSyncSource,
    plan: AddressSyncPlan,
) -> None:
    if not _sync_source_writes_remote(sync_source):
        return
    reserved_contact_ids = {item.contact_id for item in plan.items if item.contact_id and item.action != "unchanged"}
    for contact in _carddav_contacts_for_source(session, sync_source):
        if contact.id in reserved_contact_ids:
            continue
        if contact.source_ref:
            if not _local_contact_changed_after_last_sync(contact, sync_source):
                continue
            if contact.deleted_at is not None:
                if contact.source_revision:
                    _add_sync_plan_item(
                        plan,
                        AddressSyncPlanItem(
                            action="remote_delete",
                            href=contact.source_ref,
                            contact_id=contact.id,
                            display_name=contact.display_name,
                            etag=contact.source_revision,
                            message="Local delete will be pushed to CardDAV.",
                        ),
                    )
                else:
                    _add_sync_plan_item(
                        plan,
                        AddressSyncPlanItem(
                            action="conflict",
                            href=contact.source_ref,
                            contact_id=contact.id,
                            display_name=contact.display_name,
                            message="Local delete cannot be pushed because no remote ETag is known.",
                        ),
                    )
                continue
            if contact.source_revision:
                _add_sync_plan_item(
                    plan,
                    AddressSyncPlanItem(
                        action="remote_update",
                        href=contact.source_ref,
                        contact_id=contact.id,
                        display_name=contact.display_name,
                        etag=contact.source_revision,
                        raw_vcard=_carddav_contact_vcard(contact, href=contact.source_ref),
                        source_revision=contact.source_revision,
                        message="Local update will be pushed to CardDAV.",
                    ),
                )
            else:
                _add_sync_plan_item(
                    plan,
                    AddressSyncPlanItem(
                        action="conflict",
                        href=contact.source_ref,
                        contact_id=contact.id,
                        display_name=contact.display_name,
                        message="Local update cannot be pushed because no remote ETag is known.",
                    ),
                )
            continue

        if contact.deleted_at is None:
            href = _carddav_new_contact_href(sync_source, contact)
            _add_sync_plan_item(
                plan,
                AddressSyncPlanItem(
                    action="remote_create",
                    href=href,
                    contact_id=contact.id,
                    display_name=contact.display_name,
                    raw_vcard=_carddav_contact_vcard(contact, href=href),
                    message="Local contact will be created in CardDAV.",
                ),
            )


def _add_sync_plan_item(plan: AddressSyncPlan, item: AddressSyncPlanItem) -> None:
    plan.items.append(item)
    if item.action in {"create", "remote_create"}:
        plan.stats.created += 1
        plan.stats.fetched += 1
    elif item.action in {"update", "remote_update"}:
        plan.stats.updated += 1
        plan.stats.fetched += 1
    elif item.action in {"delete", "remote_delete"}:
        plan.stats.deleted += 1
    elif item.action == "conflict":
        plan.stats.conflicts += 1
        plan.stats.fetched += 1
    elif item.action == "unchanged":
        plan.stats.unchanged += 1
    elif item.action == "error":
        plan.stats.errors += 1


def _apply_address_sync_plan(session: Session, principal: ApiPrincipal, plan: AddressSyncPlan, *, client: AddressCardDAVClient | None = None) -> None:
    for item in plan.items:
        if item.action == "create":
            contact = _upsert_carddav_contact(session, principal, plan.sync_source, item)
            item.contact_id = contact.id
        elif item.action == "update":
            contact = _upsert_carddav_contact(session, principal, plan.sync_source, item)
            item.contact_id = contact.id
        elif item.action == "delete" and item.contact_id:
            contact = get_visible_contact(session, principal, item.contact_id, include_deleted=True)
            previous = _contact_change_payload(contact, prefix="previous_")
            contact.deleted_at = utcnow()
            contact.updated_by_account_id = _account_id(principal)
            record_sync_tombstone(
                session,
                principal,
                plan.sync_source.id,
                AddressSyncTombstoneCreateRequest(
                    contact_id=contact.id,
                    remote_uid=item.remote_uid,
                    resource_href=item.href,
                    local_deleted_at=contact.deleted_at,
                    remote_deleted_at=utcnow(),
                    synced_at=utcnow(),
                ),
            )
            _record_address_contact_change(session, principal, contact=contact, operation="deleted", previous=previous)
        elif item.action in REMOTE_SYNC_ACTIONS:
            if client is None:
                raise AddressBookError("CardDAV write client is required for outbound sync.")
            try:
                _apply_carddav_remote_sync_item(session, principal, plan.sync_source, item, client=client)
            except AddressCardDAVPreconditionFailed as exc:
                _record_carddav_write_conflict(session, principal, plan, item, message=str(exc))
        elif item.action == "conflict":
            local_value = {"contact_id": item.contact_id, "display_name": item.display_name}
            if item.contact_id:
                contact = get_visible_contact(session, principal, item.contact_id, include_deleted=True)
                local_value["payload"] = _contact_payload_for_conflict(contact)
            remote_payload = item.parsed_payload.model_dump(mode="json") if item.parsed_payload else None
            record_sync_conflict(
                session,
                principal,
                plan.sync_source.id,
                AddressSyncConflictCreateRequest(
                    contact_id=item.contact_id,
                    remote_uid=item.remote_uid,
                    resource_href=item.href,
                    field_path="vcard",
                    local_value=local_value,
                    remote_value={"payload": remote_payload, "display_name": item.parsed_payload.display_name if item.parsed_payload else None, "etag": item.etag},
                    metadata={"message": item.message, "raw_vcard": item.raw_vcard, "remote_payload": remote_payload, "source_revision": item.source_revision},
                ),
            )
    session.flush()


def _apply_carddav_remote_sync_item(
    session: Session,
    principal: ApiPrincipal,
    sync_source: AddressSyncSource,
    item: AddressSyncPlanItem,
    *,
    client: AddressCardDAVClient,
) -> None:
    if not item.contact_id:
        raise AddressBookError("Outbound sync item is missing a contact id.")
    contact = get_visible_contact(session, principal, item.contact_id, include_deleted=True)
    if contact.address_book_id != sync_source.address_book_id:
        raise AddressBookError("Outbound sync contact must belong to the sync source address book.")
    if item.action == "remote_create":
        href = item.href or _carddav_new_contact_href(sync_source, contact)
        raw_vcard = item.raw_vcard or _carddav_contact_vcard(contact, href=href)
        previous = _contact_change_payload(contact, prefix="previous_")
        result = client.put_object(href, raw_vcard, create=True)
        _mark_carddav_contact_synced(contact, sync_source=sync_source, href=href, raw_vcard=raw_vcard, etag=result.etag or item.source_revision or contact.source_revision)
        contact.updated_by_account_id = _account_id(principal)
        _record_address_contact_change(session, principal, contact=contact, operation="synced", previous=previous)
    elif item.action == "remote_update":
        if not item.href:
            raise AddressBookError("Outbound update is missing a CardDAV href.")
        etag = item.etag or contact.source_revision
        raw_vcard = item.raw_vcard or _carddav_contact_vcard(contact, href=item.href)
        previous = _contact_change_payload(contact, prefix="previous_")
        result = client.put_object(item.href, raw_vcard, etag=etag)
        _mark_carddav_contact_synced(contact, sync_source=sync_source, href=item.href, raw_vcard=raw_vcard, etag=result.etag or etag)
        contact.updated_by_account_id = _account_id(principal)
        _record_address_contact_change(session, principal, contact=contact, operation="synced", previous=previous)
    elif item.action == "remote_delete":
        if not item.href:
            raise AddressBookError("Outbound delete is missing a CardDAV href.")
        etag = item.etag or contact.source_revision
        previous = _contact_change_payload(contact, prefix="previous_")
        client.delete_object(item.href, etag=etag)
        record_sync_tombstone(
            session,
            principal,
            sync_source.id,
            AddressSyncTombstoneCreateRequest(
                contact_id=contact.id,
                remote_uid=item.remote_uid,
                resource_href=item.href,
                local_deleted_at=contact.deleted_at,
                remote_deleted_at=utcnow(),
                synced_at=utcnow(),
            ),
        )
        contact.updated_by_account_id = _account_id(principal)
        _record_address_contact_change(session, principal, contact=contact, operation="synced", previous=previous)


def _mark_carddav_contact_synced(
    contact: Contact,
    *,
    sync_source: AddressSyncSource,
    href: str,
    raw_vcard: str,
    etag: str | None,
) -> None:
    contact.source_kind = sync_source.connector_type
    contact.source_ref = href
    contact.source_payload_kind = "vcard"
    contact.source_payload_raw = raw_vcard
    contact.source_revision = etag
    provenance = dict(contact.provenance or {})
    provenance["carddav"] = {
        "sync_source_id": sync_source.id,
        "href": href,
        "etag": etag,
        "last_push_at": utcnow().isoformat(),
    }
    contact.provenance = provenance


def _record_carddav_write_conflict(
    session: Session,
    principal: ApiPrincipal,
    plan: AddressSyncPlan,
    item: AddressSyncPlanItem,
    *,
    message: str,
) -> None:
    original_action = item.action
    _decrement_sync_plan_stats(plan, original_action)
    item.action = "conflict"
    item.message = message
    plan.stats.conflicts += 1
    local_value: dict[str, Any] = {"contact_id": item.contact_id, "display_name": item.display_name, "action": original_action}
    if item.contact_id:
        contact = get_visible_contact(session, principal, item.contact_id, include_deleted=True)
        local_value["payload"] = _contact_payload_for_conflict(contact)
    record_sync_conflict(
        session,
        principal,
        plan.sync_source.id,
        AddressSyncConflictCreateRequest(
            contact_id=item.contact_id,
            remote_uid=item.remote_uid,
            resource_href=item.href,
            field_path="vcard",
            local_value=local_value,
            remote_value={"etag": item.etag, "message": message},
            metadata={"message": message, "source": "carddav_write"},
        ),
    )


def _decrement_sync_plan_stats(plan: AddressSyncPlan, action: str) -> None:
    if action == "remote_create":
        plan.stats.created = max(0, plan.stats.created - 1)
        plan.stats.fetched = max(0, plan.stats.fetched - 1)
    elif action == "remote_update":
        plan.stats.updated = max(0, plan.stats.updated - 1)
        plan.stats.fetched = max(0, plan.stats.fetched - 1)
    elif action == "remote_delete":
        plan.stats.deleted = max(0, plan.stats.deleted - 1)


def _upsert_carddav_contact(
    session: Session,
    principal: ApiPrincipal,
    sync_source: AddressSyncSource,
    item: AddressSyncPlanItem,
) -> Contact:
    if item.parsed_payload is None:
        raise AddressBookError("Remote vCard payload is missing.")
    contact = None
    if item.contact_id:
        contact = get_visible_contact(session, principal, item.contact_id, include_deleted=True)
    if contact is None:
        contact = Contact(
            tenant_id=sync_source.tenant_id,
            address_book_id=sync_source.address_book_id,
            display_name=_display_name_from_payload(item.parsed_payload),
            created_by_account_id=_account_id(principal),
        )
        session.add(contact)
        operation = "created"
        previous = None
    else:
        previous = _contact_change_payload(contact, prefix="previous_")
        operation = "created" if contact.deleted_at is not None else "updated"
        contact.display_name = _display_name_from_payload(item.parsed_payload, fallback=contact.display_name)
    contact.given_name = _trim(item.parsed_payload.given_name)
    contact.family_name = _trim(item.parsed_payload.family_name)
    contact.organization = _trim(item.parsed_payload.organization)
    contact.role_title = _trim(item.parsed_payload.role_title)
    contact.note = _trim(item.parsed_payload.note)
    contact.tags = _normalize_tags(item.parsed_payload.tags)
    contact.source_kind = sync_source.connector_type
    contact.source_ref = item.href
    contact.source_payload_kind = "vcard"
    contact.source_payload_raw = item.raw_vcard
    contact.source_revision = item.source_revision
    contact.provenance = {
        **(item.parsed_payload.provenance or {}),
        "carddav": {
            "sync_source_id": sync_source.id,
            "href": item.href,
            "remote_uid": item.remote_uid,
            "etag": item.etag,
        },
    }
    contact.deleted_at = None
    contact.updated_by_account_id = _account_id(principal)
    _replace_emails(contact, item.parsed_payload.emails)
    _replace_phones(contact, item.parsed_payload.phones)
    _replace_postal_addresses(contact, item.parsed_payload.postal_addresses)
    session.flush()
    _record_address_contact_change(session, principal, contact=contact, operation=operation, previous=previous)
    return contact


def _parse_single_remote_vcard(raw_vcard: str):
    result = parse_vcards_with_issues(raw_vcard)
    if not result.cards:
        message = result.issues[0].message if result.issues else "No vCard entries found."
        return None, message
    return result.cards[0], "; ".join(issue.message for issue in result.issues if issue.severity == "error") or None


def _carddav_contacts_by_href(session: Session, sync_source: AddressSyncSource) -> dict[str, Contact]:
    return {str(contact.source_ref): contact for contact in _carddav_contacts_for_source(session, sync_source) if contact.source_ref}


def _carddav_contacts_for_source(session: Session, sync_source: AddressSyncSource) -> list[Contact]:
    rows = (
        session.query(Contact)
        .filter(
            Contact.address_book_id == sync_source.address_book_id,
        )
        .all()
    )
    return rows


def _sync_source_writes_remote(sync_source: AddressSyncSource) -> bool:
    return sync_source.sync_direction in OUTBOUND_SYNC_DIRECTIONS and not sync_source.read_only


def _sync_source_reads_remote(sync_source: AddressSyncSource) -> bool:
    return sync_source.sync_direction in INBOUND_SYNC_DIRECTIONS


def _carddav_new_contact_href(sync_source: AddressSyncSource, contact: Contact) -> str:
    url = _trim(sync_source.external_address_book_ref)
    if not url:
        return f"{contact.id}.vcf"
    try:
        collection = ensure_collection_url(url)
    except AddressCardDAVError:
        return f"{contact.id}.vcf"
    path = urllib.parse.urlparse(collection).path.rstrip("/")
    if not path:
        return f"{contact.id}.vcf"
    return f"{path}/{contact.id}.vcf"


def _carddav_contact_vcard(contact: Contact, *, href: str | None = None) -> str:
    content = contacts_to_vcard([contact])
    if href and "UID:" not in content:
        content = content.replace("VERSION:4.0\r\n", f"VERSION:4.0\r\nUID:{href}\r\n", 1)
    return content


def _local_contact_changed_after_last_sync(contact: Contact, sync_source: AddressSyncSource) -> bool:
    if sync_source.last_success_at is None:
        return False
    if contact.deleted_at and contact.deleted_at > sync_source.last_success_at:
        return True
    return bool(contact.updated_at and contact.updated_at > sync_source.last_success_at)


def _carddav_client_from_payload(
    session: Session,
    principal: ApiPrincipal,
    payload: AddressCardDavDiscoveryRequest,
    *,
    source: AddressSyncSource | None = None,
) -> AddressCardDAVClient:
    _assert_no_caller_carddav_credential_ref(payload.credential_ref)
    url = payload.url or (source.external_address_book_ref if source else "")
    metadata = dict(source.metadata_ or {}) if source else {}
    auth = dict(metadata.get("carddav") or {})
    auth_type = payload.auth_type or str(auth.get("auth_type") or "none")
    username = payload.username if payload.username is not None else auth.get("username")
    password = _secret_value(payload.password)
    bearer_token = _secret_value(payload.bearer_token)
    credential_ref = payload.credential_ref if payload.credential_ref is not None else auth.get("credential_ref")
    reusable = _resolve_core_address_credential(
        session,
        tenant_id=principal.tenant_id,
        source_id=source.id if source else None,
        credential_ref=credential_ref,
    )
    username = username or _credential_username(reusable)
    secret = _resolve_carddav_secret(
        session=session,
        tenant_id=principal.tenant_id,
        source_id=source.id if source else None,
        auth_type=auth_type,
        password=password,
        bearer_token=bearer_token,
        credential_ref=credential_ref,
        encrypted=auth.get("secret_encrypted"),
    )
    return _new_carddav_client(url, auth_type=auth_type, username=username, secret=secret)


def _carddav_client_for_source(
    session: Session,
    sync_source: AddressSyncSource,
    *,
    password: str | None = None,
    bearer_token: str | None = None,
) -> AddressCardDAVClient:
    metadata = dict(sync_source.metadata_ or {})
    auth = dict(metadata.get("carddav") or {})
    auth_type = str(auth.get("auth_type") or "none")
    username = auth.get("username")
    reusable = _resolve_core_address_credential(
        session,
        tenant_id=sync_source.tenant_id or "",
        source_id=sync_source.id,
        credential_ref=auth.get("credential_ref"),
    )
    username = username or _credential_username(reusable)
    secret = _resolve_carddav_secret(
        session=session,
        tenant_id=sync_source.tenant_id or "",
        source_id=sync_source.id,
        auth_type=auth_type,
        password=password,
        bearer_token=bearer_token,
        credential_ref=auth.get("credential_ref"),
        encrypted=auth.get("secret_encrypted"),
    )
    return _new_carddav_client(sync_source.external_address_book_ref or "", auth_type=auth_type, username=username, secret=secret)


def _new_carddav_client(url: str, *, auth_type: str, username: str | None, secret: str | None) -> AddressCardDAVClient:
    if auth_type == "basic":
        if not username:
            raise AddressBookError("CardDAV basic auth requires a username.")
        if not secret:
            raise AddressBookError("CardDAV basic auth requires a password or credential reference.")
        return AddressCardDAVClient(collection_url=url, username=username, password=secret)
    if auth_type == "bearer":
        if not secret:
            raise AddressBookError("CardDAV bearer auth requires a token or credential reference.")
        return AddressCardDAVClient(collection_url=url, bearer_token=secret)
    return AddressCardDAVClient(collection_url=url)


def _carddav_metadata(
    *,
    auth_type: str,
    username: str | None,
    password: str | None,
    bearer_token: str | None,
    credential_ref: str | None,
    collection_url: str,
) -> dict[str, Any]:
    secret = password if auth_type == "basic" else bearer_token if auth_type == "bearer" else None
    auth: dict[str, Any] = {
        "auth_type": auth_type,
        "username": _trim(username),
        "credential_ref": _trim(credential_ref),
    }
    if secret:
        auth["secret_encrypted"] = encrypt_secret(secret)
    return {
        "carddav": auth,
        "collection_url": collection_url,
    }


def _resolve_carddav_secret(
    *,
    session: Session,
    tenant_id: str,
    source_id: str | None,
    auth_type: str,
    password: str | None,
    bearer_token: str | None,
    credential_ref: str | None,
    encrypted: str | None,
) -> str | None:
    if auth_type == "basic" and password:
        return password
    if auth_type == "bearer" and bearer_token:
        return bearer_token
    if credential_ref:
        reusable = _resolve_core_address_credential(
            session,
            tenant_id=tenant_id,
            source_id=source_id,
            credential_ref=credential_ref,
        )
        if reusable is None:
            raise AddressBookError(
                "The CardDAV credential reference is not a server-owned credential or visible reusable credential envelope."
            )
        return _credential_secret(reusable, auth_type=auth_type)
    if encrypted:
        return decrypt_secret(encrypted)
    return None


def _core_credential_id(credential_ref: str | None) -> str | None:
    if not credential_ref or not credential_ref.startswith(CORE_CREDENTIAL_ENVELOPE_PREFIX):
        return None
    value = credential_ref.removeprefix(CORE_CREDENTIAL_ENVELOPE_PREFIX).strip()
    return value or None


def _resolve_core_address_credential(
    session: Session,
    *,
    tenant_id: str,
    source_id: str | None,
    credential_ref: str | None,
) -> ResolvedCredentialEnvelope | None:
    credential_id = _core_credential_id(credential_ref)
    if credential_id is None:
        return None
    try:
        return resolve_credential_envelope(
            session,
            credential_id=credential_id,
            context=address_credential_context(
                tenant_id=tenant_id,
                source_id=source_id,
            ),
        )
    except CredentialEnvelopeError as exc:
        raise AddressBookError(
            "The selected reusable credential is unavailable to this CardDAV source."
        ) from exc


def _credential_username(credential: ResolvedCredentialEnvelope | None) -> str | None:
    if credential is None:
        return None
    value = credential.public_data.get("username")
    return _trim(value)


def _credential_secret(credential: ResolvedCredentialEnvelope, *, auth_type: str) -> str | None:
    keys = (
        ("password", "secret", "token")
        if auth_type == "basic"
        else ("access_token", "bearer_token", "token", "password", "secret")
    )
    for key in keys:
        value = credential.secret_data.get(key)
        if value is not None and str(value):
            return str(value)
    return None


def resolve_trusted_deployment_carddav_credential_ref(credential_ref: str) -> str | None:
    """Resolve env-backed credentials only for trusted deployment code.

    API-managed discovery and sync-source paths deliberately never call this
    function, so a tenant user cannot select arbitrary process environment
    variables as connector credentials.
    """

    if not credential_ref.startswith(CARDDAV_SECRET_ENV_PREFIX):
        raise AddressBookError("Trusted deployment credential references must use the env: prefix")
    env_name = credential_ref.removeprefix(CARDDAV_SECRET_ENV_PREFIX)
    if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", env_name):
        raise AddressBookError("Trusted deployment credential reference contains an invalid environment variable name")
    return os.environ.get(env_name)


def public_address_sync_metadata(metadata: object) -> dict[str, Any]:
    payload = copy.deepcopy(metadata) if isinstance(metadata, dict) else {}
    auth = payload.get("carddav")
    if not isinstance(auth, dict):
        return payload
    had_credential = bool(auth.get("secret_encrypted") or auth.get("credential_ref"))
    auth["credential_envelope_id"] = _core_credential_id(auth.get("credential_ref"))
    auth.pop("secret_encrypted", None)
    auth.pop("credential_ref", None)
    auth["has_credential"] = had_credential
    return payload


def _assert_no_caller_carddav_credential_ref(credential_ref: str | None) -> None:
    value = _trim(credential_ref)
    if value and not value.startswith(CORE_CREDENTIAL_ENVELOPE_PREFIX):
        raise AddressBookError(
            "Caller-supplied credential references are accepted only for reusable credential envelopes."
        )


def _carddav_auth_metadata(metadata: object) -> dict[str, Any]:
    if not isinstance(metadata, dict):
        return {}
    carddav = metadata.get("carddav")
    return carddav if isinstance(carddav, dict) else {}


def _assert_api_carddav_metadata_safe(metadata: object) -> None:
    auth = _carddav_auth_metadata(metadata)
    if auth.get("credential_ref") or auth.get("secret_encrypted"):
        raise AddressBookError(
            "CardDAV credential references and encrypted secrets are server-managed; provide credentials through the CardDAV source endpoint"
        )


def _merge_server_owned_carddav_metadata(existing: object, incoming: dict[str, Any]) -> dict[str, Any]:
    merged = copy.deepcopy(incoming)
    existing_auth = _carddav_auth_metadata(existing)
    server_owned = {
        key: existing_auth[key]
        for key in ("credential_ref", "secret_encrypted")
        if existing_auth.get(key)
    }
    if not server_owned:
        return merged
    auth = merged.get("carddav")
    if not isinstance(auth, dict):
        auth = {}
        merged["carddav"] = auth
    auth.update(server_owned)
    return merged


def _secret_value(value: Any | None) -> str | None:
    if value is None:
        return None
    if hasattr(value, "get_secret_value"):
        return str(value.get_secret_value())
    return str(value)


def _sync_plan_diagnostic(plan: AddressSyncPlan) -> dict[str, Any]:
    return {
        "created": plan.stats.created,
        "updated": plan.stats.updated,
        "deleted": plan.stats.deleted,
        "conflicts": plan.stats.conflicts,
        "unchanged": plan.stats.unchanged,
        "errors": plan.stats.errors,
        "full_sync": plan.stats.full_sync,
        "used_sync_token": plan.stats.used_sync_token,
    }


def _contact_change_payload(contact: Contact, *, prefix: str = "") -> dict[str, Any]:
    return {
        f"{prefix}address_book_id": contact.address_book_id,
        f"{prefix}display_name": contact.display_name,
        f"{prefix}source_kind": contact.source_kind,
        f"{prefix}source_ref": contact.source_ref,
        f"{prefix}source_revision": contact.source_revision,
    }


def _contact_payload_for_conflict(contact: Contact) -> dict[str, Any]:
    return {
        "display_name": contact.display_name,
        "given_name": contact.given_name,
        "family_name": contact.family_name,
        "organization": contact.organization,
        "role_title": contact.role_title,
        "note": contact.note,
        "tags": list(contact.tags or []),
        "emails": [
            {"label": email.label, "email": email.email, "is_primary": email.is_primary}
            for email in sorted(contact.emails, key=lambda item: (item.order_index, item.id))
        ],
        "phones": [
            {"label": phone.label, "phone": phone.phone, "is_primary": phone.is_primary}
            for phone in sorted(contact.phones, key=lambda item: (item.order_index, item.id))
        ],
        "postal_addresses": [
            {
                "label": address.label,
                "street": address.street,
                "postal_code": address.postal_code,
                "locality": address.locality,
                "region": address.region,
                "country": address.country,
                "is_primary": address.is_primary,
            }
            for address in sorted(contact.postal_addresses, key=lambda item: (item.order_index, item.id))
        ],
    }


def _record_address_contact_change(
    session: Session,
    principal: ApiPrincipal,
    *,
    contact: Contact,
    operation: str,
    previous: dict[str, Any] | None = None,
) -> None:
    payload = _contact_change_payload(contact)
    if previous:
        payload.update(previous)
    record_change(
        session,
        module_id=ADDRESS_MODULE_ID,
        collection=ADDRESS_CONTACTS_COLLECTION,
        resource_type=ADDRESS_CONTACT_RESOURCE,
        resource_id=contact.id,
        operation=operation,
        tenant_id=contact.tenant_id,
        actor_type="user",
        actor_id=_account_id(principal),
        payload=payload,
    )


def _read_only_from_sync_direction(sync_direction: str, explicit_read_only: bool | None) -> bool:
    if sync_direction in READ_ONLY_SYNC_DIRECTIONS:
        return True
    if explicit_read_only is not None:
        return explicit_read_only
    return False


def _apply_sync_source_to_book(book: AddressBook, sync_source: AddressSyncSource) -> None:
    book.source_kind = sync_source.connector_type
    book.source_ref = sync_source.id
    book.read_only = sync_source.read_only or sync_source.sync_direction in READ_ONLY_SYNC_DIRECTIONS
    book.sync_status = sync_source.status
    book.sync_error = sync_source.last_error
    book.updated_by_account_id = sync_source.updated_by_account_id


def _sync_contact_or_none(session: Session, principal: ApiPrincipal, sync_source: AddressSyncSource, contact_id: str | None) -> Contact | None:
    if not contact_id:
        return None
    contact = get_visible_contact(session, principal, contact_id, include_deleted=True)
    if contact.address_book_id != sync_source.address_book_id:
        raise AddressBookError("Sync contact must belong to the sync source address book.")
    return contact


def _visible_address_lists_query(session: Session, principal: ApiPrincipal, *, include_deleted: bool = False, include_deleted_books: bool = False):
    book_ids = [book.id for book in list_address_books(session, principal, include_deleted=include_deleted_books)]
    if not book_ids:
        return session.query(AddressList).filter(false())
    query = session.query(AddressList).filter(AddressList.address_book_id.in_(book_ids))
    if not include_deleted:
        query = query.filter(AddressList.deleted_at.is_(None))
    return query


def list_address_lists(
    session: Session,
    principal: ApiPrincipal,
    *,
    address_book_id: str | None = None,
    include_deleted: bool = False,
) -> list[AddressList]:
    query = _visible_address_lists_query(session, principal, include_deleted=include_deleted, include_deleted_books=include_deleted)
    if address_book_id:
        get_visible_address_book(session, principal, address_book_id, include_deleted=include_deleted)
        query = query.filter(AddressList.address_book_id == address_book_id)
    return query.order_by(AddressList.name.asc(), AddressList.id.asc()).all()


def get_visible_address_list(session: Session, principal: ApiPrincipal, address_list_id: str, *, include_deleted: bool = False) -> AddressList:
    address_list = (
        _visible_address_lists_query(session, principal, include_deleted=include_deleted, include_deleted_books=include_deleted)
        .filter(AddressList.id == address_list_id)
        .one_or_none()
    )
    if address_list is None:
        raise AddressBookError("Address list not found.")
    return address_list


def create_address_list(session: Session, principal: ApiPrincipal, address_book_id: str, payload: AddressListCreateRequest) -> AddressList:
    book = get_visible_address_book(session, principal, address_book_id)
    _require_mutable_book(book)
    name = _trim(payload.name)
    if not name:
        raise AddressBookError("Address list name is required.")
    address_list = AddressList(
        tenant_id=book.tenant_id,
        address_book_id=book.id,
        name=name,
        description=_trim(payload.description),
        source_kind=book.source_kind,
        read_only=book.read_only,
        created_by_account_id=_account_id(principal),
        updated_by_account_id=_account_id(principal),
        metadata_={},
    )
    session.add(address_list)
    return address_list


def update_address_list(session: Session, principal: ApiPrincipal, address_list_id: str, payload: AddressListUpdateRequest) -> AddressList:
    address_list = get_visible_address_list(session, principal, address_list_id)
    _require_mutable_address_list(address_list)
    if "name" in payload.model_fields_set:
        name = _trim(payload.name)
        if not name:
            raise AddressBookError("Address list name is required.")
        address_list.name = name
    if "description" in payload.model_fields_set:
        address_list.description = _trim(payload.description)
    address_list.updated_by_account_id = _account_id(principal)
    return address_list


def delete_address_list(session: Session, principal: ApiPrincipal, address_list_id: str) -> None:
    address_list = get_visible_address_list(session, principal, address_list_id)
    _require_mutable_address_list(address_list)
    address_list.deleted_at = utcnow()
    address_list.updated_by_account_id = _account_id(principal)


def restore_address_list(session: Session, principal: ApiPrincipal, address_list_id: str) -> AddressList:
    address_list = get_visible_address_list(session, principal, address_list_id, include_deleted=True)
    _require_mutable_book(address_list.address_book)
    if address_list.read_only:
        raise AddressBookError("Address list is read-only.")
    address_list.deleted_at = None
    address_list.updated_by_account_id = _account_id(principal)
    return address_list


def list_address_list_entries(session: Session, principal: ApiPrincipal, address_list_id: str) -> list[AddressListEntry]:
    address_list = get_visible_address_list(session, principal, address_list_id)
    return (
        session.query(AddressListEntry)
        .join(Contact, AddressListEntry.contact_id == Contact.id)
        .filter(AddressListEntry.address_list_id == address_list.id, Contact.deleted_at.is_(None))
        .order_by(AddressListEntry.order_index.asc(), AddressListEntry.id.asc())
        .all()
    )


def create_address_list_entry(
    session: Session,
    principal: ApiPrincipal,
    address_list_id: str,
    payload: AddressListEntryCreateRequest,
) -> AddressListEntry:
    address_list = get_visible_address_list(session, principal, address_list_id)
    _require_mutable_address_list(address_list)
    contact = get_visible_contact(session, principal, payload.contact_id)
    if contact.address_book_id != address_list.address_book_id:
        raise AddressBookError("Address-list entries must reference contacts in the same address book.")
    contact_email = _contact_email_by_id(contact, payload.contact_email_id)
    postal_address = _contact_postal_address_by_id(contact, payload.contact_postal_address_id)
    if contact_email is not None and postal_address is not None:
        raise AddressBookError("Address-list entries may reference either an email address or a postal address, not both.")
    order_index = _next_address_list_entry_order(session, address_list.id)
    entry = AddressListEntry(
        address_list_id=address_list.id,
        contact_id=contact.id,
        contact_email_id=contact_email.id if contact_email is not None else None,
        contact_postal_address_id=postal_address.id if postal_address is not None else None,
        target_kind=_address_list_entry_kind(contact_email, postal_address),
        label=_trim(payload.label),
        order_index=order_index,
        metadata_={},
    )
    address_list.updated_by_account_id = _account_id(principal)
    session.add(entry)
    return entry


def delete_address_list_entry(session: Session, principal: ApiPrincipal, entry_id: str) -> None:
    entry = get_visible_address_list_entry(session, principal, entry_id)
    _require_mutable_address_list(entry.address_list)
    entry.address_list.updated_by_account_id = _account_id(principal)
    session.delete(entry)


def get_visible_address_list_entry(session: Session, principal: ApiPrincipal, entry_id: str) -> AddressListEntry:
    list_ids = [address_list.id for address_list in list_address_lists(session, principal)]
    if not list_ids:
        raise AddressBookError("Address-list entry not found.")
    entry = session.query(AddressListEntry).filter(AddressListEntry.id == entry_id, AddressListEntry.address_list_id.in_(list_ids)).one_or_none()
    if entry is None:
        raise AddressBookError("Address-list entry not found.")
    return entry


def _contact_email_by_id(contact: Contact, email_id: str | None) -> ContactEmail | None:
    if not email_id:
        return None
    email = next((item for item in contact.emails if item.id == email_id), None)
    if email is None:
        raise AddressBookError("Contact email not found.")
    return email


def _contact_postal_address_by_id(contact: Contact, postal_address_id: str | None) -> ContactPostalAddress | None:
    if not postal_address_id:
        return None
    postal_address = next((item for item in contact.postal_addresses if item.id == postal_address_id), None)
    if postal_address is None:
        raise AddressBookError("Contact postal address not found.")
    return postal_address


def _address_list_entry_kind(contact_email: ContactEmail | None, postal_address: ContactPostalAddress | None) -> str:
    if contact_email is not None:
        return "email"
    if postal_address is not None:
        return "postal_address"
    return "contact"


def _next_address_list_entry_order(session: Session, address_list_id: str) -> int:
    current = session.query(func.max(AddressListEntry.order_index)).filter(AddressListEntry.address_list_id == address_list_id).scalar()
    return int(current or 0) + 1


def create_address_book(
    session: Session,
    principal: ApiPrincipal,
    payload: AddressBookCreateRequest,
    *,
    allow_system: bool = False,
) -> AddressBook:
    name = _trim(payload.name)
    if not name:
        raise AddressBookError("Address book name is required.")
    tenant_id, scope_id = _scope_id_for_create(principal, payload, allow_system=allow_system)
    book = AddressBook(
        tenant_id=tenant_id,
        scope_type=payload.scope_type,
        scope_id=scope_id,
        name=name,
        description=_trim(payload.description),
        source_kind="local",
        read_only=False,
        created_by_account_id=_account_id(principal),
        updated_by_account_id=_account_id(principal),
        metadata_={},
    )
    session.add(book)
    return book


def update_address_book(session: Session, principal: ApiPrincipal, book_id: str, payload: AddressBookUpdateRequest) -> AddressBook:
    book = get_visible_address_book(session, principal, book_id)
    _require_mutable_book(book)
    if "name" in payload.model_fields_set:
        name = _trim(payload.name)
        if not name:
            raise AddressBookError("Address book name is required.")
        book.name = name
    if "description" in payload.model_fields_set:
        book.description = _trim(payload.description)
    book.updated_by_account_id = _account_id(principal)
    return book


def delete_address_book(session: Session, principal: ApiPrincipal, book_id: str) -> None:
    book = get_visible_address_book(session, principal, book_id)
    _require_mutable_book(book)
    deleted_at = utcnow()
    book.deleted_at = deleted_at
    book.updated_by_account_id = _account_id(principal)
    for contact in book.contacts:
        if contact.deleted_at is None:
            contact.deleted_at = deleted_at
            contact.updated_by_account_id = _account_id(principal)


def restore_address_book(session: Session, principal: ApiPrincipal, book_id: str, *, restore_contacts: bool = True) -> AddressBook:
    book = get_visible_address_book(session, principal, book_id, include_deleted=True)
    if book.read_only:
        raise AddressBookError("Address book is read-only.")
    book.deleted_at = None
    book.updated_by_account_id = _account_id(principal)
    if restore_contacts:
        for contact in book.contacts:
            if contact.deleted_at is not None:
                contact.deleted_at = None
                contact.updated_by_account_id = _account_id(principal)
    return book


def _primary_email(payloads: list[ContactEmailPayload]) -> str | None:
    if not payloads:
        return None
    primary = next((item.email for item in payloads if item.is_primary), None)
    return _trim(primary) or _trim(payloads[0].email)


def _display_name_from_payload(payload: ContactCreateRequest | ContactUpdateRequest, *, fallback: str | None = None) -> str:
    explicit = _trim(payload.display_name)
    if explicit:
        return explicit
    parts = [_trim(payload.given_name), _trim(payload.family_name)]
    joined = " ".join(part for part in parts if part)
    if joined:
        return joined
    emails = getattr(payload, "emails", None)
    if isinstance(emails, list):
        email = _primary_email(emails)
        if email:
            return email
    if fallback:
        return fallback
    raise AddressBookError("Contact name or email is required.")


def _normalize_tags(tags: list[str] | None) -> list[str]:
    seen: set[str] = set()
    normalized: list[str] = []
    for tag in tags or []:
        value = tag.strip()
        key = value.casefold()
        if value and key not in seen:
            seen.add(key)
            normalized.append(value)
    return normalized


def _replace_emails(contact: Contact, payloads: list[ContactEmailPayload]) -> None:
    contact.emails.clear()
    normalized = [_trim(item.email) for item in payloads]
    normalized_payloads = [(item, email) for item, email in zip(payloads, normalized, strict=False) if email]
    primary_index = next((index for index, (item, _email) in enumerate(normalized_payloads) if item.is_primary), 0)
    for index, (item, email) in enumerate(normalized_payloads):
        contact.emails.append(
            ContactEmail(
                label=_trim(item.label),
                email=email or "",
                is_primary=index == primary_index,
                order_index=index,
            )
        )


def _replace_phones(contact: Contact, payloads: list[ContactPhonePayload]) -> None:
    contact.phones.clear()
    normalized = [_trim(item.phone) for item in payloads]
    normalized_payloads = [(item, phone) for item, phone in zip(payloads, normalized, strict=False) if phone]
    primary_index = next((index for index, (item, _phone) in enumerate(normalized_payloads) if item.is_primary), 0)
    for index, (item, phone) in enumerate(normalized_payloads):
        contact.phones.append(
            ContactPhone(
                label=_trim(item.label),
                phone=phone or "",
                is_primary=index == primary_index,
                order_index=index,
            )
        )


def _postal_payload_has_content(payload: ContactPostalAddressPayload) -> bool:
    return any(_trim(value) for value in (payload.street, payload.postal_code, payload.locality, payload.region, payload.country))


def _replace_postal_addresses(contact: Contact, payloads: list[ContactPostalAddressPayload]) -> None:
    contact.postal_addresses.clear()
    normalized_payloads = [item for item in payloads if _postal_payload_has_content(item)]
    primary_index = next((index for index, item in enumerate(normalized_payloads) if item.is_primary), 0)
    for index, item in enumerate(normalized_payloads):
        contact.postal_addresses.append(
            ContactPostalAddress(
                label=_trim(item.label),
                street=_trim(item.street),
                postal_code=_trim(item.postal_code),
                locality=_trim(item.locality),
                region=_trim(item.region),
                country=_trim(item.country),
                is_primary=index == primary_index,
                order_index=index,
            )
        )


def create_contact(session: Session, principal: ApiPrincipal, address_book_id: str, payload: ContactCreateRequest) -> Contact:
    book = get_visible_address_book(session, principal, address_book_id)
    _require_mutable_book(book)
    contact = Contact(
        tenant_id=book.tenant_id,
        address_book_id=book.id,
        display_name=_display_name_from_payload(payload),
        given_name=_trim(payload.given_name),
        family_name=_trim(payload.family_name),
        organization=_trim(payload.organization),
        role_title=_trim(payload.role_title),
        note=_trim(payload.note),
        tags=_normalize_tags(payload.tags),
        source_kind=book.source_kind,
        provenance=payload.provenance or {},
        created_by_account_id=_account_id(principal),
        updated_by_account_id=_account_id(principal),
        metadata_={},
    )
    _replace_emails(contact, payload.emails)
    _replace_phones(contact, payload.phones)
    _replace_postal_addresses(contact, payload.postal_addresses)
    session.add(contact)
    return contact


def import_vcards(session: Session, principal: ApiPrincipal, address_book_id: str, content: str) -> VCardImportResult:
    book = get_visible_address_book(session, principal, address_book_id)
    _require_mutable_book(book)
    contacts: list[Contact] = []
    result = parse_vcards_with_issues(content)
    for parsed in result.cards:
        contact = create_contact(session, principal, book.id, parsed.payload)
        contact.source_kind = "vcard"
        contact.source_ref = _trim(parsed.source_ref)
        contact.source_payload_kind = "vcard"
        contact.source_payload_raw = parsed.raw
        contact.source_revision = _trim(parsed.source_revision)
        contacts.append(contact)
    return VCardImportResult(contacts=contacts, issues=result.issues, skipped=result.skipped)


def _visible_contact_query(session: Session, principal: ApiPrincipal, *, include_deleted: bool = False, include_deleted_books: bool = False):
    book_ids = [book.id for book in list_address_books(session, principal, include_deleted=include_deleted_books)]
    if not book_ids:
        return session.query(Contact).filter(false())
    query = session.query(Contact).filter(Contact.address_book_id.in_(book_ids))
    if not include_deleted:
        query = query.filter(Contact.deleted_at.is_(None))
    return query


def list_contacts(
    session: Session,
    principal: ApiPrincipal,
    *,
    address_book_id: str | None = None,
    query: str | None = None,
    limit: int = 200,
    include_deleted: bool = False,
) -> list[Contact]:
    contact_query = _visible_contact_query(session, principal, include_deleted=include_deleted, include_deleted_books=include_deleted)
    if address_book_id:
        get_visible_address_book(session, principal, address_book_id, include_deleted=include_deleted)
        contact_query = contact_query.filter(Contact.address_book_id == address_book_id)
    normalized_query = _trim(query)
    if normalized_query:
        pattern = f"%{normalized_query.lower()}%"
        contact_query = contact_query.outerjoin(ContactEmail).filter(
            or_(
                func.lower(Contact.display_name).like(pattern),
                func.lower(Contact.organization).like(pattern),
                func.lower(ContactEmail.email).like(pattern),
            )
        )
    return contact_query.order_by(Contact.display_name.asc(), Contact.id.asc()).limit(max(1, min(limit, 500))).all()


def get_visible_contact(session: Session, principal: ApiPrincipal, contact_id: str, *, include_deleted: bool = False) -> Contact:
    contact = _visible_contact_query(session, principal, include_deleted=include_deleted, include_deleted_books=include_deleted).filter(Contact.id == contact_id).one_or_none()
    if contact is None:
        raise AddressBookError("Contact not found.")
    return contact


def update_contact(session: Session, principal: ApiPrincipal, contact_id: str, payload: ContactUpdateRequest) -> Contact:
    contact = get_visible_contact(session, principal, contact_id)
    _require_mutable_book(contact.address_book)
    fallback_name = contact.display_name
    if any(field in payload.model_fields_set for field in ("display_name", "given_name", "family_name", "emails")):
        contact.display_name = _display_name_from_payload(payload, fallback=fallback_name)
    for field_name, attr_name in (
        ("given_name", "given_name"),
        ("family_name", "family_name"),
        ("organization", "organization"),
        ("role_title", "role_title"),
        ("note", "note"),
    ):
        if field_name in payload.model_fields_set:
            setattr(contact, attr_name, _trim(getattr(payload, field_name)))
    if "tags" in payload.model_fields_set:
        contact.tags = _normalize_tags(payload.tags)
    if "provenance" in payload.model_fields_set:
        contact.provenance = payload.provenance or {}
    if "emails" in payload.model_fields_set:
        _replace_emails(contact, payload.emails or [])
    if "phones" in payload.model_fields_set:
        _replace_phones(contact, payload.phones or [])
    if "postal_addresses" in payload.model_fields_set:
        _replace_postal_addresses(contact, payload.postal_addresses or [])
    contact.updated_by_account_id = _account_id(principal)
    return contact


def delete_contact(session: Session, principal: ApiPrincipal, contact_id: str) -> None:
    contact = get_visible_contact(session, principal, contact_id)
    _require_mutable_book(contact.address_book)
    contact.deleted_at = utcnow()
    contact.updated_by_account_id = _account_id(principal)


def restore_contact(session: Session, principal: ApiPrincipal, contact_id: str) -> Contact:
    contact = get_visible_contact(session, principal, contact_id, include_deleted=True)
    _require_mutable_book(contact.address_book)
    contact.deleted_at = None
    contact.updated_by_account_id = _account_id(principal)
    return contact


def export_address_book_vcard(session: Session, principal: ApiPrincipal, address_book_id: str) -> tuple[AddressBook, str]:
    book = get_visible_address_book(session, principal, address_book_id)
    contacts = (
        session.query(Contact)
        .filter(Contact.address_book_id == book.id, Contact.deleted_at.is_(None))
        .order_by(Contact.display_name.asc(), Contact.id.asc())
        .all()
    )
    return book, contacts_to_vcard(contacts)


def export_contact_vcard(session: Session, principal: ApiPrincipal, contact_id: str) -> tuple[Contact, str]:
    contact = get_visible_contact(session, principal, contact_id)
    return contact, contacts_to_vcard([contact])