govoplan-mail/src/govoplan_mail/backend/router.py

from __future__ import annotations

import dataclasses
import hashlib
from types import SimpleNamespace
from typing import Any

from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy import func, or_, select
from sqlalchemy.orm import Session

from govoplan_mail.backend.schemas import (
    MailAddressLookupCandidate,
    MailAddressLookupResponse,
    MailAddressWriteTarget,
    MailAddressWriteTargetResponse,
    MailContactCreateRequest,
    MailContactCreateResponse,
    MailConnectionTestResponse,
    MailBounceObservationListResponse,
    MailBounceObservationResponse,
    MailBounceScanResponse,
    MailBounceSourceListResponse,
    MailBounceSourceRequest,
    MailBounceSourceResponse,
    MailCampaignCredentialCreateRequest,
    MailCredentialBindRequest,
    MailCredentialCreateRequest,
    MailCredentialEnvelopeResponse,
    MailCredentialListResponse,
    MailCredentialUpdateRequest,
    MailImapFolderListResponse,
    MailImapFolderResponse,
    MailDeliveryCommandResponse,
    MailDeliveryReconcileRequest,
    MailDeliveryResendRequest,
    MailMailboxBootstrapResponse,
    MailMailboxChangesResponse,
    MailImapTestRequest,
    MailMailboxAttachmentResponse,
    MailMailboxMessageDetailResponse,
    MailMailboxMessageListResponse,
    MailMailboxMessageResponse,
    MailMailboxMessageSummaryResponse,
    MailProfilePolicyResponse,
    MailProfilePolicyUpdateRequest,
    MailSettingsDeltaResponse,
    MailPop3ImportListResponse,
    MailPop3ImportRecordResponse,
    MailPop3ImportRequest,
    MailPop3ImportResponse,
    MailPop3MessagePreviewResponse,
    MailPop3PreviewRequest,
    MailPop3PreviewResponse,
    MailServerProfileCreateRequest,
    MailServerEndpointCreateRequest,
    MailServerEndpointResponse,
    MailServerEndpointUpdateRequest,
    MailServerProfileListResponse,
    MailServerProfileResponse,
    MailServerProfileUpdateRequest,
    MailSmtpTestRequest,
)
from govoplan_core.auth import ApiPrincipal, get_api_principal, has_scope
from govoplan_core.audit.logging import audit_event
from govoplan_core.api.v1.schemas import DeltaDeletedItem
from govoplan_core.core.change_sequence import (
    ChangeSequenceEntry,
    decode_sequence_watermark,
    encode_sequence_watermark,
    record_change,
    sequence_watermark_is_expired,
)
from govoplan_core.core.pagination import KeysetCursorError, decode_keyset_cursor, encode_keyset_cursor, keyset_query_fingerprint
from govoplan_core.db.session import get_session
from govoplan_mail.backend.mailbox_index import (
    cache_mailbox_folders,
    cache_mailbox_messages,
    cached_mailbox_folders,
    cached_mailbox_message_page,
    clear_mailbox_index,
)
from govoplan_mail.backend.mail_profiles import (
    MailProfileError,
    assert_mail_policy_allows_transport,
    campaign_mail_owner_context,
    campaign_mail_context_visible_to_actor,
    create_mail_server_profile,
    delete_mail_profile_credentials,
    effective_mail_profile_policy_for_scope,
    get_mail_profile_policy,
    get_mail_server_profile,
    get_mail_server_profile_for_actor,
    imap_config_from_profile,
    list_mail_server_profiles,
    mail_profile_visible_to_actor,
    mail_profile_scope_visible_to_actor,
    parent_mail_profile_policy,
    profile_response_payload,
    set_mail_profile_policy,
    smtp_config_from_profile,
    update_mail_server_profile,
)
from govoplan_mail.backend.config import ImapConfig, JmapConfig, Pop3Config, SmtpConfig
from govoplan_mail.backend.db.models import MailPop3Import
from govoplan_mail.backend.pop3_imports import (
    Pop3ImportError,
    create_pop3_imports,
    list_pop3_imports,
    mark_pop3_deletion_result,
    pop3_import_payload,
)
from govoplan_mail.backend.runtime import get_registry
from govoplan_mail.backend.recovery import (
    MailRecoveryError,
    MailboxRefreshBusy,
    begin_mailbox_refresh_recovery,
)
from govoplan_mail.backend.delivery_outbox import (
    MailDeliveryError,
    delivery_command_diagnostics,
    delivery_command_summary,
    get_delivery_command,
    reconcile_delivery_command,
    resend_delivery_command,
)
from govoplan_mail.backend.bounce_processing import (
    MailBounceError,
    SqlMailBounceProcessingProvider,
    configure_bounce_source,
    delete_bounce_source,
    list_bounce_observations,
    list_bounce_sources,
)
from govoplan_mail.backend.server_hierarchy import (
    MailHierarchyContext,
    MailServerHierarchyError,
    bind_mail_credential,
    create_and_bind_mail_credential,
    create_mail_server_endpoint,
    get_available_mail_credential,
    get_bound_mail_credential,
    get_mail_server_endpoint,
    hierarchy_context_for_profile,
    initialize_profile_hierarchy,
    list_available_mail_credentials,
    mail_credential_payload,
    mail_server_ref,
    mail_server_endpoint_payload,
    profile_hierarchy_payload,
    resolve_mail_transport,
    sync_default_profile_credential,
    sync_default_profile_server,
    unlink_mail_credential,
    update_bound_mail_credential,
    update_mail_server_endpoint,
)
from govoplan_mail.backend.sending.imap import ImapAppendError, ImapConfigurationError, get_imap_message, list_imap_folders, list_imap_messages, load_imap_mailbox_bootstrap, test_imap_login
from govoplan_mail.backend.sending.jmap import (
    JmapAuthenticationError,
    JmapCapabilityError,
    JmapConfigurationError,
    JmapPermissionError,
    JmapProviderError,
    get_jmap_email_changes,
    get_jmap_message,
    list_jmap_folders,
    list_jmap_messages,
    load_jmap_mailbox_bootstrap,
    test_jmap_connection,
)
from govoplan_mail.backend.sending.smtp import test_smtp_login
from govoplan_mail.backend.sending.pop3 import (
    Pop3ConfigurationError,
    Pop3ProviderError,
    delete_pop3_messages,
    download_pop3_messages,
    preview_pop3_messages,
    test_pop3_login,
)

router = APIRouter(prefix="/mail", tags=["mail"])

MAIL_MODULE_ID = "mail"
MAIL_PROFILES_COLLECTION = "mail.profiles"
MAIL_POLICIES_COLLECTION = "mail.profile_policies"
MAIL_SETTINGS_COLLECTIONS = (MAIL_PROFILES_COLLECTION, MAIL_POLICIES_COLLECTION)
MAIL_PROFILE_RESOURCE = "mail_profile"
MAIL_POLICY_RESOURCE = "mail_profile_policy"
MAIL_SERVER_RESOURCE = "mail_server"
MAIL_CREDENTIAL_RESOURCE = "mail_credential"
MAILBOX_MESSAGES_CURSOR_SCOPE = "mail.mailbox.messages.v1"
JMAP_MAILBOX_MESSAGES_CURSOR_SCOPE = "mail.mailbox.jmap.messages.v1"
DEFAULT_MAILBOX_MESSAGE_LIMIT = 50
CAPABILITY_ADDRESSES_LOOKUP = "addresses.lookup"
CAPABILITY_ADDRESSES_CONTACT_WRITER = "addresses.contact_writer"
bounce_provider = SqlMailBounceProcessingProvider()


@router.get(
    "/delivery-commands/{command_id}",
    response_model=MailDeliveryCommandResponse,
)
def get_mail_delivery_command(
    command_id: str,
    session: Session = Depends(get_session),
    principal: ApiPrincipal = Depends(get_api_principal),
):
    """Return the business-safe status of one tenant Mail command."""

    _require_scope(principal, "mail:profile:read")
    try:
        command = get_delivery_command(
            session,
            tenant_id=principal.tenant_id,
            command_id=command_id,
        )
    except MailDeliveryError as exc:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=str(exc),
        ) from exc
    return MailDeliveryCommandResponse(result=delivery_command_summary(command))


@router.get(
    "/delivery-commands/{command_id}/diagnostics",
    response_model=MailDeliveryCommandResponse,
)
def get_mail_delivery_command_diagnostics(
    command_id: str,
    session: Session = Depends(get_session),
    principal: ApiPrincipal = Depends(get_api_principal),
):
    """Return bounded recipient and attempt evidence to diagnostic actors."""

    _require_scope(principal, "mail:delivery:diagnostic")
    try:
        result = delivery_command_diagnostics(
            session,
            tenant_id=principal.tenant_id,
            command_id=command_id,
        )
    except MailDeliveryError as exc:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=str(exc),
        ) from exc
    return MailDeliveryCommandResponse(result=result)


@router.post(
    "/delivery-commands/{command_id}/reconcile",
    response_model=MailDeliveryCommandResponse,
)
def reconcile_mail_delivery_command(
    command_id: str,
    payload: MailDeliveryReconcileRequest,
    session: Session = Depends(get_session),
    principal: ApiPrincipal = Depends(get_api_principal),
):
    _require_scope(principal, "mail:delivery:reconcile")
    try:
        result = reconcile_delivery_command(
            session,
            tenant_id=principal.tenant_id,
            command_id=command_id,
            decision=payload.decision,
            evidence_reference=payload.evidence_reference,
            note=payload.note,
            user_id=principal.user.id,
        )
    except MailDeliveryError as exc:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=str(exc),
        ) from exc
    return MailDeliveryCommandResponse(result=result)


@router.post(
    "/delivery-commands/{command_id}/resend",
    response_model=MailDeliveryCommandResponse,
)
def resend_mail_delivery_command(
    command_id: str,
    payload: MailDeliveryResendRequest,
    session: Session = Depends(get_session),
    principal: ApiPrincipal = Depends(get_api_principal),
):
    _require_scope(principal, "mail:delivery:reconcile")
    try:
        result = resend_delivery_command(
            session,
            tenant_id=principal.tenant_id,
            command_id=command_id,
            idempotency_key=payload.idempotency_key,
            user_id=principal.user.id,
        )
    except MailDeliveryError as exc:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=str(exc),
        ) from exc
    return MailDeliveryCommandResponse(result=result)


@router.get("/bounce-sources", response_model=MailBounceSourceListResponse)
def get_mail_bounce_sources(
    session: Session = Depends(get_session),
    principal: ApiPrincipal = Depends(get_api_principal),
) -> MailBounceSourceListResponse:
    _require_scope(principal, "mail:bounce:read")
    return MailBounceSourceListResponse(
        sources=[
            MailBounceSourceResponse.model_validate(item, from_attributes=True)
            for item in list_bounce_sources(
                session,
                tenant_id=principal.tenant_id,
            )
        ]
    )


@router.post("/bounce-sources", response_model=MailBounceSourceResponse)
def save_mail_bounce_source(
    payload: MailBounceSourceRequest,
    session: Session = Depends(get_session),
    principal: ApiPrincipal = Depends(get_api_principal),
) -> MailBounceSourceResponse:
    _require_scope(principal, "mail:bounce:manage")
    try:
        source = configure_bounce_source(
            session,
            tenant_id=principal.tenant_id,
            profile_id=payload.profile_id,
            folder=payload.folder,
            imap_server_id=payload.imap_server_id,
            imap_credential_id=payload.imap_credential_id,
            is_active=payload.is_active,
            created_by_user_id=principal.user.id,
        )
    except MailBounceError as exc:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail=str(exc),
        ) from exc
    audit_event(
        session,
        tenant_id=principal.tenant_id,
        user_id=principal.user.id,
        action="mail.bounce_source.saved",
        object_type="mail_bounce_source",
        object_id=source.id,
        details={
            "profile_id": source.profile_id,
            "folder": source.folder,
            "is_active": source.is_active,
        },
    )
    session.commit()
    return MailBounceSourceResponse.model_validate(source, from_attributes=True)


@router.delete("/bounce-sources/{source_id}", status_code=status.HTTP_204_NO_CONTENT)
def remove_mail_bounce_source(
    source_id: str,
    session: Session = Depends(get_session),
    principal: ApiPrincipal = Depends(get_api_principal),
) -> None:
    _require_scope(principal, "mail:bounce:manage")
    try:
        delete_bounce_source(
            session,
            tenant_id=principal.tenant_id,
            source_id=source_id,
        )
    except MailBounceError as exc:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
    audit_event(
        session,
        tenant_id=principal.tenant_id,
        user_id=principal.user.id,
        action="mail.bounce_source.deleted",
        object_type="mail_bounce_source",
        object_id=source_id,
        details={},
    )
    session.commit()


@router.post(
    "/bounce-sources/{source_id}/scan",
    response_model=MailBounceScanResponse,
)
def scan_mail_bounce_source(
    source_id: str,
    limit: int = Query(default=100, ge=1, le=1_000),
    session: Session = Depends(get_session),
    principal: ApiPrincipal = Depends(get_api_principal),
) -> MailBounceScanResponse:
    _require_scope(principal, "mail:bounce:manage")
    try:
        result = bounce_provider.scan_source(
            session,
            tenant_id=principal.tenant_id,
            source_id=source_id,
            limit=limit,
        )
    except MailBounceError as exc:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail=str(exc),
        ) from exc
    session.commit()
    return MailBounceScanResponse.model_validate(result)


@router.get(
    "/bounce-observations",
    response_model=MailBounceObservationListResponse,
)
def get_mail_bounce_observations(
    command_id: str | None = Query(default=None, max_length=36),
    limit: int = Query(default=100, ge=1, le=500),
    session: Session = Depends(get_session),
    principal: ApiPrincipal = Depends(get_api_principal),
) -> MailBounceObservationListResponse:
    _require_scope(principal, "mail:bounce:read")
    return MailBounceObservationListResponse(
        observations=[
            MailBounceObservationResponse.model_validate(
                dataclasses.asdict(item)
            )
            for item in list_bounce_observations(
                session,
                tenant_id=principal.tenant_id,
                command_id=command_id,
                limit=limit,
            )
        ]
    )


def _capability_payload(value: object) -> dict[str, Any]:
    if dataclasses.is_dataclass(value):
        return dataclasses.asdict(value)
    if isinstance(value, dict):
        return dict(value)
    payload: dict[str, Any] = {}
    for key in (
        "contact_id",
        "address_book_id",
        "display_name",
        "email",
        "email_label",
        "organization",
        "role_title",
        "tags",
        "source_kind",
        "source_ref",
        "source_revision",
        "provenance",
        "address_book_label",
        "operation",
        "allowed",
        "reason",
        "message",
        "scope_type",
        "scope_id",
        "read_only",
        "required_scopes",
    ):
        if hasattr(value, key):
            payload[key] = getattr(value, key)
    return payload


def _registry_capability(name: str) -> object | None:
    registry = get_registry()
    if registry is None or not hasattr(registry, "has_capability") or not registry.has_capability(name):
        return None
    return registry.capability(name)


def _require_scope(principal: ApiPrincipal, scope: str) -> None:
    if not has_scope(principal, scope):
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=f"Missing scope: {scope}")


def _require_any_scope(principal: ApiPrincipal, *scopes: str) -> None:
    if not any(has_scope(principal, scope) for scope in scopes):
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Requires one of: " + ", ".join(scopes))


def _is_own_user_scope(principal: ApiPrincipal, scope_type: str, scope_id: str | None) -> bool:
    return scope_type == "user" and scope_id == principal.user.id


def _require_profile_write_scope(
    principal: ApiPrincipal,
    scope_type: str,
    scope_id: str | None = None,
) -> None:
    if scope_type == "system":
        _require_scope(principal, "system:settings:write")
        return
    if has_scope(principal, "mail:profile:write"):
        return
    if _is_own_user_scope(principal, scope_type, scope_id) and has_scope(
        principal, "mail:profile:write_own"
    ):
        return
    raise HTTPException(
        status_code=status.HTTP_403_FORBIDDEN,
        detail="Missing scope: mail:profile:write",
    )


def _require_profile_credentials_scope(
    principal: ApiPrincipal,
    scope_type: str,
    scope_id: str | None = None,
) -> None:
    if scope_type == "system":
        _require_scope(principal, "system:settings:write")
        return
    if has_scope(principal, "mail:secret:manage"):
        return
    if _is_own_user_scope(principal, scope_type, scope_id) and has_scope(
        principal, "mail:secret:manage_own"
    ):
        return
    raise HTTPException(
        status_code=status.HTTP_403_FORBIDDEN,
        detail="Missing scope: mail:secret:manage",
    )


def _profile_for_mutation(
    session: Session,
    *,
    principal: ApiPrincipal,
    profile_id: str,
):
    """Resolve a profile through the actor's mutation boundary.

    Mutation routes deliberately do not apply effective transport policy here:
    an owner must still be able to repair or deactivate a profile after policy
    becomes more restrictive.  Missing and non-owned identifiers are kept
    indistinguishable for self-service actors.
    """

    try:
        profile = get_mail_server_profile(
            session,
            tenant_id=principal.tenant_id,
            profile_id=profile_id,
            for_update=True,
            mutation_owner_user_id=(
                principal.user.id
                if has_scope(principal, "mail:profile:write_own")
                else None
            ),
            can_manage_tenant_profiles=_principal_can_manage_tenant_profiles(principal),
            can_manage_system_profiles=_principal_can_manage_system_profiles(principal),
        )
        _require_profile_write_scope(
            principal,
            profile.scope_type or "tenant",
            profile.scope_id,
        )
    except (MailProfileError, HTTPException) as exc:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Mail-server profile not found",
        ) from exc
    return profile


def _server_for_mutation(
    session: Session,
    *,
    principal: ApiPrincipal,
    profile_id: str,
    server_id: str,
):
    profile = _profile_for_mutation(
        session,
        principal=principal,
        profile_id=profile_id,
    )
    try:
        server = get_mail_server_endpoint(
            session,
            profile=profile,
            server_id=server_id,
            context=hierarchy_context_for_profile(
                profile,
                user_id=principal.user.id,
                group_ids=principal.group_ids,
                administrative=True,
            ),
            require_active=False,
            for_update=True,
        )
    except MailServerHierarchyError as exc:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Mail server not found",
        ) from exc
    return profile, server


def _server_response(
    session: Session,
    *,
    principal: ApiPrincipal,
    profile,
    server,
) -> MailServerEndpointResponse:
    context = hierarchy_context_for_profile(
        profile,
        user_id=principal.user.id,
        group_ids=principal.group_ids,
        administrative=True,
    )
    return MailServerEndpointResponse.model_validate(
        mail_server_endpoint_payload(
            session,
            server=server,
            context=context,
            include_inactive_credentials=True,
        )
    )


def _record_profile_child_change(
    session: Session,
    *,
    principal: ApiPrincipal,
    profile,
    operation: str,
    resource_type: str,
    resource_id: str,
) -> None:
    _record_mail_change(
        session,
        collection=MAIL_PROFILES_COLLECTION,
        resource_type=MAIL_PROFILE_RESOURCE,
        resource_id=profile.id,
        operation="updated",
        principal=principal,
        tenant_id=profile.tenant_id,
        payload={
            "scope_type": profile.scope_type,
            "scope_id": profile.scope_id,
            "child_operation": operation,
            "child_resource_type": resource_type,
            "child_resource_id": resource_id,
        },
    )


_TRANSPORT_ENDPOINT_FIELDS = ("host", "port", "security")


def _transport_endpoint_changed(
    current: dict[str, Any] | None,
    updated: SmtpConfig | ImapConfig,
) -> bool:
    current_payload = current or {}
    updated_payload = updated.model_dump(mode="json")
    return any(
        current_payload.get(field) != updated_payload.get(field)
        for field in _TRANSPORT_ENDPOINT_FIELDS
    )


def _profile_update_requires_credentials_scope(
    profile,
    *,
    payload: MailServerProfileUpdateRequest,
    smtp: SmtpConfig | None,
    imap: ImapConfig | None,
) -> bool:
    if payload.credentials_supplied() or payload.clear_imap:
        return True
    if (
        smtp is not None
        and getattr(profile, "smtp_password_encrypted", None)
        and _transport_endpoint_changed(profile.smtp_config, smtp)
    ):
        return True
    return bool(
        imap is not None
        and getattr(profile, "imap_password_encrypted", None)
        and _transport_endpoint_changed(profile.imap_config, imap)
    )


def _profile_has_stored_credentials(profile) -> bool:
    return bool(
        getattr(profile, "smtp_password_encrypted", None)
        or getattr(profile, "imap_password_encrypted", None)
    )


def _require_mailbox_read_scope(principal: ApiPrincipal) -> None:
    _require_scope(principal, "mail:mailbox:read")
    _require_scope(principal, "mail:profile:use")


def _principal_is_tenant_admin(principal: ApiPrincipal) -> bool:
    return has_scope(principal, "tenant:*")


def _principal_can_manage_tenant_profiles(principal: ApiPrincipal) -> bool:
    return has_scope(principal, "mail:profile:write")


def _principal_can_manage_own_profiles(principal: ApiPrincipal) -> bool:
    return has_scope(principal, "mail:profile:write_own")


def _principal_can_manage_system_profiles(principal: ApiPrincipal) -> bool:
    return has_scope(principal, "system:settings:write")


def _profile_actor_kwargs(
    principal: ApiPrincipal,
    *,
    administrative_visibility: bool,
) -> dict[str, object]:
    return {
        "actor_user_id": principal.user.id,
        "actor_group_ids": principal.group_ids,
        "actor_can_manage_own_profiles": _principal_can_manage_own_profiles(principal),
        "actor_can_manage_tenant_profiles": _principal_can_manage_tenant_profiles(principal),
        "actor_can_manage_system_profiles": _principal_can_manage_system_profiles(principal),
        "actor_tenant_admin": _principal_is_tenant_admin(principal),
        "actor_administrative_visibility": administrative_visibility,
    }


def _get_profile_for_principal(
    session: Session,
    *,
    principal: ApiPrincipal,
    profile_id: str,
    require_active: bool = False,
    administrative_visibility: bool = False,
):
    return get_mail_server_profile_for_actor(
        session,
        tenant_id=principal.tenant_id,
        profile_id=profile_id,
        user_id=principal.user.id,
        group_ids=principal.group_ids,
        can_manage_own_profiles=_principal_can_manage_own_profiles(principal),
        can_manage_tenant_profiles=_principal_can_manage_tenant_profiles(principal),
        can_manage_system_profiles=_principal_can_manage_system_profiles(principal),
        tenant_admin=_principal_is_tenant_admin(principal),
        administrative_visibility=administrative_visibility,
        require_active=require_active,
    )


def _require_policy_read_scope(principal: ApiPrincipal, scope_type: str) -> None:
    if scope_type == "system":
        _require_scope(principal, "system:settings:read")
    elif scope_type == "tenant":
        _require_any_scope(principal, "admin:policies:read", "mail:profile:read")
    else:
        _require_any_scope(principal, "admin:policies:read", "mail:profile:read", "campaigns:campaign:read")


def _require_policy_write_scope(principal: ApiPrincipal, scope_type: str) -> None:
    if scope_type == "system":
        _require_scope(principal, "system:settings:write")
    elif scope_type == "tenant":
        _require_scope(principal, "admin:policies:write")
    else:
        _require_any_scope(principal, "admin:policies:write", "mail:profile:write")


def _require_policy_scope_visibility(
    session: Session,
    *,
    principal: ApiPrincipal,
    scope_type: str,
    scope_id: str | None,
) -> None:
    if scope_type in {"system", "tenant"}:
        return
    if has_scope(principal, "admin:policies:read"):
        return
    if not mail_profile_scope_visible_to_actor(
        session,
        scope_type=scope_type,
        scope_id=scope_id,
        profile_tenant_id=principal.tenant_id,
        tenant_id=principal.tenant_id,
        user_id=principal.user.id,
        group_ids=principal.group_ids,
        can_manage_tenant_profiles=_principal_can_manage_tenant_profiles(principal),
        tenant_admin=_principal_is_tenant_admin(principal),
    ):
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Mail profile policy not found")


def _require_campaign_context_visibility(
    session: Session,
    *,
    principal: ApiPrincipal,
    campaign_id: str | None,
) -> None:
    if not campaign_id:
        return
    if not campaign_mail_context_visible_to_actor(
        session,
        tenant_id=principal.tenant_id,
        campaign_id=campaign_id,
        user_id=principal.user.id,
        group_ids=principal.group_ids,
        tenant_admin=_principal_is_tenant_admin(principal),
    ):
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Campaign not found")


def _require_policy_campaign_context(
    session: Session,
    *,
    principal: ApiPrincipal,
    scope_type: str,
    scope_id: str | None,
    campaign_id: str | None,
) -> None:
    if scope_type == "campaign" and campaign_id and campaign_id != scope_id:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail="campaign_id must match the campaign policy scope_id",
        )
    _require_campaign_context_visibility(
        session,
        principal=principal,
        campaign_id=campaign_id or (scope_id if scope_type == "campaign" else None),
    )


def _require_campaign_profile_mutation_access(
    session: Session,
    *,
    principal: ApiPrincipal,
    scope_type: str,
    scope_id: str | None,
) -> None:
    if scope_type != "campaign":
        return
    _require_campaign_context_visibility(
        session,
        principal=principal,
        campaign_id=scope_id,
    )


def _mail_hierarchy_context(
    principal: ApiPrincipal,
    *,
    target_scope_type: str | None = None,
    target_scope_id: str | None = None,
    administrative: bool = False,
) -> MailHierarchyContext:
    resolved_scope_type = target_scope_type or "user"
    resolved_scope_id = (
        target_scope_id
        if target_scope_type is not None
        else principal.user.id
    )
    return MailHierarchyContext(
        tenant_id=principal.tenant_id,
        user_id=principal.user.id,
        group_ids=frozenset(str(item) for item in principal.group_ids),
        target_scope_type=resolved_scope_type,
        target_scope_id=resolved_scope_id,
        administrative=administrative,
    )


def _profile_response(
    profile,
    *,
    session: Session | None = None,
    principal: ApiPrincipal | None = None,
    target_scope_type: str | None = None,
    target_scope_id: str | None = None,
    include_inactive: bool = False,
    administrative: bool = False,
) -> MailServerProfileResponse:
    payload = profile_response_payload(profile)
    if session is None or principal is None:
        return MailServerProfileResponse.model_validate(payload)
    context = _mail_hierarchy_context(
        principal,
        target_scope_type=target_scope_type,
        target_scope_id=target_scope_id,
        administrative=administrative,
    )
    servers = profile_hierarchy_payload(
        session,
        profile=profile,
        context=context,
        include_inactive=include_inactive,
    )
    payload["servers"] = servers
    credentials: dict[str, dict[str, Any]] = {
        "smtp": {"username": None},
        "imap": {"username": None},
    }
    for protocol in ("smtp", "imap"):
        candidates = [server for server in servers if server["protocol"] == protocol]
        selected = next(
            (
                server
                for server in candidates
                if server["is_default"] and server["is_active"]
            ),
            candidates[0] if candidates else None,
        )
        if selected is None:
            continue
        payload[protocol] = dict(selected["config"])
        bound_credentials = list(selected.get("credentials") or [])
        selected_credential = next(
            (
                credential
                for credential in bound_credentials
                if credential.get("is_default") and credential.get("is_active")
            ),
            bound_credentials[0] if bound_credentials else None,
        )
        if selected_credential is not None:
            credentials[protocol] = {
                "username": (
                    selected_credential.get("public_data") or {}
                ).get("username")
            }
        payload[f"{protocol}_password_configured"] = any(
            credential.get("secret_configured")
            for server in candidates
            for credential in (server.get("credentials") or [])
        )
    payload["credentials"] = credentials
    return MailServerProfileResponse.model_validate(payload)


def _policy_response(
    session: Session,
    *,
    tenant_id: str,
    scope_type: str,
    scope_id: str | None,
    campaign_id: str | None = None,
) -> MailProfilePolicyResponse:
    policy = get_mail_profile_policy(session, tenant_id=tenant_id, scope_type=scope_type, scope_id=scope_id)
    effective_policy = effective_mail_profile_policy_for_scope(
        session,
        tenant_id=tenant_id,
        scope_type=scope_type,
        scope_id=scope_id,
        campaign_id=campaign_id,
    )
    effective = effective_policy.as_dict()
    effective_sources = effective_policy.source_policies
    parent_policy = parent_mail_profile_policy(session, tenant_id=tenant_id, scope_type=scope_type, scope_id=scope_id) if scope_type != "system" else None
    parent = parent_policy.as_dict() if parent_policy else None
    parent_sources = parent_policy.source_policies if parent_policy else []
    return MailProfilePolicyResponse(
        scope_type=scope_type,
        scope_id=scope_id,
        policy=policy,
        effective_policy=effective,
        parent_policy=parent,
        effective_policy_sources=effective_sources,
        parent_policy_sources=parent_sources,
    )


def _record_mail_change(
    session: Session,
    *,
    collection: str,
    resource_type: str,
    resource_id: str | None,
    operation: str,
    principal: ApiPrincipal,
    tenant_id: str | None,
    payload: dict[str, object] | None = None,
) -> None:
    if not resource_id:
        return
    record_change(
        session,
        module_id=MAIL_MODULE_ID,
        collection=collection,
        resource_type=resource_type,
        resource_id=resource_id,
        operation=operation,
        tenant_id=tenant_id,
        actor_type="user",
        actor_id=principal.user.id,
        payload=payload or {},
    )


def _mail_delta_query(session: Session, *, tenant_id: str, since_sequence: int):
    return session.query(ChangeSequenceEntry).filter(
        ChangeSequenceEntry.id > since_sequence,
        ChangeSequenceEntry.module_id == MAIL_MODULE_ID,
        ChangeSequenceEntry.collection.in_(MAIL_SETTINGS_COLLECTIONS),
        or_(ChangeSequenceEntry.tenant_id == tenant_id, ChangeSequenceEntry.tenant_id.is_(None)),
    )


def _mail_settings_watermark(session: Session, *, tenant_id: str) -> str:
    sequence_id = _mail_delta_query(session, tenant_id=tenant_id, since_sequence=0).with_entities(func.max(ChangeSequenceEntry.id)).scalar()
    return encode_sequence_watermark(int(sequence_id or 0))


def _mail_settings_entries(session: Session, *, tenant_id: str, since: str, limit: int):
    try:
        since_sequence = decode_sequence_watermark(since)
    except ValueError as exc:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
    expired_for_tenant = sequence_watermark_is_expired(
        session,
        since=since_sequence,
        tenant_id=tenant_id,
        module_id=MAIL_MODULE_ID,
        collections=MAIL_SETTINGS_COLLECTIONS,
    )
    expired_for_system = sequence_watermark_is_expired(
        session,
        since=since_sequence,
        tenant_id=None,
        module_id=MAIL_MODULE_ID,
        collections=MAIL_SETTINGS_COLLECTIONS,
    )
    if expired_for_tenant or expired_for_system:
        return None, False
    entries_plus_one = _mail_delta_query(
        session,
        tenant_id=tenant_id,
        since_sequence=since_sequence,
    ).order_by(ChangeSequenceEntry.id.asc()).limit(limit + 1).all()
    has_more = len(entries_plus_one) > limit
    return entries_plus_one[:limit], has_more


def _mail_settings_response_watermark(session: Session, *, tenant_id: str, entries, has_more: bool) -> str:
    return encode_sequence_watermark(entries[-1].id) if has_more and entries else _mail_settings_watermark(session, tenant_id=tenant_id)


def _mail_deleted_item(entry) -> DeltaDeletedItem:
    return DeltaDeletedItem(
        id=entry.resource_id,
        resource_type=entry.resource_type,
        revision=encode_sequence_watermark(entry.id),
        deleted_at=entry.created_at if entry.operation == "deleted" else None,
    )


def _profile_change_visible_to_principal(
    session: Session,
    *,
    entry: ChangeSequenceEntry,
    principal: ApiPrincipal,
) -> bool:
    try:
        profile = get_mail_server_profile(
            session,
            tenant_id=principal.tenant_id,
            profile_id=entry.resource_id,
        )
    except MailProfileError:
        payload = entry.payload if isinstance(entry.payload, dict) else {}
        return mail_profile_scope_visible_to_actor(
            session,
            scope_type=str(payload.get("scope_type") or "tenant"),
            scope_id=str(payload["scope_id"]) if payload.get("scope_id") else None,
            profile_tenant_id=entry.tenant_id,
            tenant_id=principal.tenant_id,
            user_id=principal.user.id,
            group_ids=principal.group_ids,
            can_manage_tenant_profiles=_principal_can_manage_tenant_profiles(principal),
            tenant_admin=_principal_is_tenant_admin(principal),
        )
    return mail_profile_visible_to_actor(
        session,
        profile=profile,
        tenant_id=principal.tenant_id,
        user_id=principal.user.id,
        group_ids=principal.group_ids,
        can_manage_own_profiles=_principal_can_manage_own_profiles(principal),
        can_manage_tenant_profiles=_principal_can_manage_tenant_profiles(principal),
        can_manage_system_profiles=_principal_can_manage_system_profiles(principal),
        tenant_admin=_principal_is_tenant_admin(principal),
        administrative_visibility=True,
        require_active=False,
    )


def _mailbox_summary_response(message) -> MailMailboxMessageSummaryResponse:
    return MailMailboxMessageSummaryResponse(
        uid=message.uid,
        folder=message.folder,
        subject=message.subject,
        from_header=message.from_header,
        to_header=message.to_header,
        cc_header=message.cc_header,
        date=message.date,
        message_id=message.message_id,
        flags=message.flags,
        size_bytes=message.size_bytes,
        body_preview=message.body_preview,
        attachment_count=message.attachment_count,
    )


def _mailbox_detail_response(message) -> MailMailboxMessageDetailResponse:
    return MailMailboxMessageDetailResponse(
        uid=message.uid,
        folder=message.folder,
        subject=message.subject,
        from_header=message.from_header,
        to_header=message.to_header,
        cc_header=message.cc_header,
        date=message.date,
        message_id=message.message_id,
        flags=message.flags,
        size_bytes=message.size_bytes,
        body_preview=message.body_preview,
        body_text=message.body_text,
        body_html=message.body_html,
        headers=message.headers,
        attachments=[MailMailboxAttachmentResponse(filename=item.filename, content_type=item.content_type, size_bytes=item.size_bytes) for item in message.attachments],
    )


def _mailbox_cursor_fingerprint(*, tenant_id: str, profile_id: str, folder: str, limit: int) -> str:
    return keyset_query_fingerprint(
        MAILBOX_MESSAGES_CURSOR_SCOPE,
        {"tenant_id": tenant_id, "profile_id": profile_id, "folder": folder, "limit": limit, "sort": "imap.uid.desc"},
    )


def _mailbox_cursor_values(cursor: str | None, *, fingerprint: str | None = None) -> dict[str, object] | None:
    try:
        return decode_keyset_cursor(MAILBOX_MESSAGES_CURSOR_SCOPE, cursor, fingerprint=fingerprint)
    except KeysetCursorError as exc:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc


def _mailbox_cursor_limit(cursor: str | None, explicit_limit: int | None) -> int:
    if explicit_limit is not None:
        return explicit_limit
    values = _mailbox_cursor_values(cursor) if cursor else None
    raw_limit = values.get("limit") if values else DEFAULT_MAILBOX_MESSAGE_LIMIT
    if not isinstance(raw_limit, int) or raw_limit < 1 or raw_limit > 100:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid pagination cursor")
    return raw_limit


def _mailbox_cursor_position(cursor: str | None, *, fingerprint: str, offset: int) -> tuple[int, str | None, str | None, dict[str, object] | None]:
    values = _mailbox_cursor_values(cursor, fingerprint=fingerprint) if cursor else None
    if values is None:
        return offset, None, None, None
    raw_offset = values.get("offset")
    if not isinstance(raw_offset, int) or raw_offset < 0:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid pagination cursor")
    after_uid = values.get("after_uid") or values.get("anchor_uid")
    uidvalidity = values.get("uidvalidity")
    if not isinstance(after_uid, str) or not after_uid:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid pagination cursor")
    if uidvalidity is not None and not isinstance(uidvalidity, str):
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid pagination cursor")
    return raw_offset, after_uid, uidvalidity, values


def _next_mailbox_cursor(
    *,
    tenant_id: str,
    profile_id: str,
    folder: str,
    limit: int,
    offset: int,
    total_count: int,
    uidvalidity: str | None,
    messages,
) -> tuple[str | None, bool]:
    cursor_stable = bool(uidvalidity)
    next_offset = offset + len(messages)
    if not cursor_stable or next_offset >= total_count or not messages:
        return None, cursor_stable
    return (
        encode_keyset_cursor(
            MAILBOX_MESSAGES_CURSOR_SCOPE,
            fingerprint=_mailbox_cursor_fingerprint(tenant_id=tenant_id, profile_id=profile_id, folder=folder, limit=limit),
            values={
                "offset": next_offset,
                "limit": limit,
                "uidvalidity": uidvalidity,
                "after_uid": messages[-1].uid,
            },
        ),
        cursor_stable,
    )


def _jmap_mailbox_cursor_fingerprint(
    *,
    tenant_id: str,
    profile_id: str,
    folder: str,
    limit: int,
    query: str,
) -> str:
    return keyset_query_fingerprint(
        JMAP_MAILBOX_MESSAGES_CURSOR_SCOPE,
        {
            "tenant_id": tenant_id,
            "profile_id": profile_id,
            "folder": folder,
            "limit": limit,
            "query": query,
            "protocol": "jmap",
        }
    )


def _jmap_mailbox_cursor_position(
    cursor: str | None,
    *,
    fingerprint: str,
    offset: int,
) -> tuple[int, str | None]:
    if not cursor:
        return offset, None
    try:
        values = decode_keyset_cursor(
            JMAP_MAILBOX_MESSAGES_CURSOR_SCOPE,
            cursor,
            fingerprint=fingerprint,
        )
    except KeysetCursorError as exc:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=str(exc),
        ) from exc
    raw_offset = values.get("offset")
    query_state = values.get("query_state")
    if not isinstance(raw_offset, int) or raw_offset < 0:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid JMAP pagination cursor")
    if not isinstance(query_state, str) or not query_state:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid JMAP pagination cursor")
    return raw_offset, query_state


def _jmap_mailbox_cursor_limit(cursor: str | None, explicit_limit: int | None) -> int:
    if explicit_limit is not None:
        return explicit_limit
    if not cursor:
        return DEFAULT_MAILBOX_MESSAGE_LIMIT
    try:
        values = decode_keyset_cursor(JMAP_MAILBOX_MESSAGES_CURSOR_SCOPE, cursor)
    except KeysetCursorError as exc:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
    raw_limit = values.get("limit")
    if not isinstance(raw_limit, int) or raw_limit < 1 or raw_limit > 100:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid JMAP pagination cursor")
    return raw_limit


def _next_jmap_mailbox_cursor(
    *,
    tenant_id: str,
    profile_id: str,
    folder: str,
    limit: int,
    offset: int,
    total_count: int,
    query_state: str,
    messages,
    query: str,
) -> tuple[str | None, bool]:
    next_offset = offset + len(messages)
    if next_offset >= total_count or not messages:
        return None, True
    return (
        encode_keyset_cursor(
            JMAP_MAILBOX_MESSAGES_CURSOR_SCOPE,
            fingerprint=_jmap_mailbox_cursor_fingerprint(
                tenant_id=tenant_id,
                profile_id=profile_id,
                folder=folder,
                limit=limit,
                query=query,
            ),
            values={
                "offset": next_offset,
                "limit": limit,
                "query_state": query_state,
            },
        ),
        True,
    )


def _mailbox_folder_response(
    result,
    *,
    from_cache: bool = False,
    refreshing: bool = False,
    indexed_at=None,
) -> MailImapFolderListResponse:
    folders = [MailImapFolderResponse(name=item.name, flags=item.flags, message_count=item.message_count, unseen_count=item.unseen_count) for item in result.folders]
    protocol = getattr(result, "protocol", "imap")
    return MailImapFolderListResponse(
        ok=True,
        protocol=protocol,
        host=result.host,
        port=result.port,
        security=result.security,
        message=f"Found {len(folders)} {protocol.upper()} folder(s).",
        folders=folders,
        detected_sent_folder=result.detected_sent_folder,
        detected_folder_mappings=getattr(result, "detected_folder_mappings", None) or {},
        from_cache=from_cache,
        refreshing=refreshing,
        indexed_at=indexed_at,
    )


def _mailbox_messages_response(
    *,
    profile_id: str,
    folder: str,
    host: str | None,
    port: int | None,
    security: str | None,
    total_count: int,
    offset: int,
    limit: int,
    cursor: str | None,
    next_cursor: str | None,
    cursor_stable: bool,
    full: bool,
    messages,
    from_cache: bool = False,
    refreshing: bool = False,
    indexed_at=None,
) -> MailMailboxMessageListResponse:
    return MailMailboxMessageListResponse(
        profile_id=profile_id,
        folder=folder,
        host=host,
        port=port,
        security=security,
        total_count=total_count,
        offset=offset,
        limit=limit,
        cursor=cursor,
        next_cursor=next_cursor,
        cursor_stable=cursor_stable,
        full=full,
        from_cache=from_cache,
        refreshing=refreshing,
        indexed_at=indexed_at,
        messages=[_mailbox_summary_response(message) for message in messages],
    )


def _cached_folder_selection(folders, requested: str, detected_sent_folder: str | None = None) -> str:
    names = {folder.name for folder in folders}
    if requested in names:
        return requested
    if "INBOX" in names:
        return "INBOX"
    if detected_sent_folder and detected_sent_folder in names:
        return detected_sent_folder
    return folders[0].name if folders else requested


def _transport_context_for_principal(
    session: Session,
    *,
    principal: ApiPrincipal,
    campaign_id: str | None = None,
) -> MailHierarchyContext:
    if not campaign_id:
        return _mail_hierarchy_context(principal)
    _require_campaign_context_visibility(
        session,
        principal=principal,
        campaign_id=campaign_id,
    )
    campaign = campaign_mail_owner_context(
        session,
        tenant_id=principal.tenant_id,
        campaign_id=campaign_id,
    )
    return MailHierarchyContext(
        tenant_id=principal.tenant_id,
        user_id=campaign.owner_user_id,
        group_ids=(
            frozenset({campaign.owner_group_id})
            if campaign.owner_group_id
            else frozenset()
        ),
        target_scope_type="campaign",
        target_scope_id=campaign.id,
    )


def _imap_config_for_principal(
    session: Session,
    *,
    principal: ApiPrincipal,
    profile_id: str,
    server_id: str | None = None,
    credential_id: str | None = None,
    campaign_id: str | None = None,
):
    profile = _get_profile_for_principal(
        session,
        principal=principal,
        profile_id=profile_id,
        require_active=True,
    )
    try:
        return resolve_mail_transport(
            session,
            profile=profile,
            protocol="imap",
            context=_transport_context_for_principal(
                session,
                principal=principal,
                campaign_id=campaign_id,
            ),
            server_id=server_id,
            credential_id=credential_id,
        ).config
    except MailServerHierarchyError as exc:
        raise MailProfileError(str(exc)) from exc


def _jmap_config_for_principal(
    session: Session,
    *,
    principal: ApiPrincipal,
    profile_id: str,
    server_id: str | None = None,
    credential_id: str | None = None,
    campaign_id: str | None = None,
) -> JmapConfig:
    profile = _get_profile_for_principal(
        session,
        principal=principal,
        profile_id=profile_id,
        require_active=True,
    )
    try:
        resolved = resolve_mail_transport(
            session,
            profile=profile,
            protocol="jmap",
            context=_transport_context_for_principal(
                session,
                principal=principal,
                campaign_id=campaign_id,
            ),
            server_id=server_id,
            credential_id=credential_id,
        )
    except MailServerHierarchyError as exc:
        raise MailProfileError(str(exc)) from exc
    if not isinstance(resolved.config, JmapConfig):
        raise MailProfileError("Mail-server profile has no JMAP configuration")
    _assert_jmap_policy_allows_config(
        session,
        principal=principal,
        profile=profile,
        config=resolved.config,
        campaign_id=campaign_id,
    )
    return resolved.config


def _assert_jmap_policy_allows_config(
    session: Session,
    *,
    principal: ApiPrincipal,
    profile,
    config: JmapConfig | dict[str, Any],
    campaign_id: str | None = None,
) -> None:
    session_url = config.session_url if isinstance(config, JmapConfig) else config.get("session_url")
    scope_type = profile.scope_type or "tenant"
    scope_id = profile.scope_id
    policy_context: dict[str, str | None] = {}
    if campaign_id or scope_type == "campaign":
        policy_context["campaign_id"] = campaign_id or scope_id
    elif scope_type == "user":
        policy_context["owner_user_id"] = scope_id
    elif scope_type == "group":
        policy_context["owner_group_id"] = scope_id
    assert_mail_policy_allows_transport(
        session,
        tenant_id=principal.tenant_id,
        smtp=None,
        jmap={"session_url": session_url},
        **policy_context,
    )


def _require_mailbox_protocol(value: str) -> str:
    clean = value.strip().casefold() if isinstance(value, str) else "imap"
    if clean not in {"imap", "jmap"}:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail="Mailbox protocol must be imap or jmap",
        )
    return clean


def _parent_mail_profile_policy_payload(session: Session, *, tenant_id: str, scope_type: str, scope_id: str | None):
    if scope_type == "system":
        return None
    return parent_mail_profile_policy(session, tenant_id=tenant_id, scope_type=scope_type, scope_id=scope_id).as_dict()


def _full_mail_settings_delta_response(
    session: Session,
    *,
    principal: ApiPrincipal,
    scope_type: str,
    scope_id: str | None,
    include_inactive: bool,
    campaign_id: str | None,
) -> MailSettingsDeltaResponse:
    profiles = list_mail_server_profiles(
        session,
        tenant_id=principal.tenant_id,
        include_inactive=include_inactive,
        campaign_id=campaign_id,
        **_profile_actor_kwargs(principal, administrative_visibility=True),
    )
    return MailSettingsDeltaResponse(
        profiles=[
            _profile_response(
                profile,
                session=session,
                principal=principal,
                target_scope_type=scope_type,
                target_scope_id=scope_id,
                include_inactive=include_inactive,
                administrative=True,
            )
            for profile in profiles
        ],
        policy=_policy_response(session, tenant_id=principal.tenant_id, scope_type=scope_type, scope_id=scope_id, campaign_id=campaign_id),
        changed_sections=["profiles", "policy"],
        deleted=[],
        watermark=_mail_settings_watermark(session, tenant_id=principal.tenant_id),
        has_more=False,
        full=True,
    )


@router.get("/address-lookup", response_model=MailAddressLookupResponse)
def lookup_mail_addresses(
    query: str = Query(min_length=1),
    limit: int = Query(default=25, ge=1, le=100),
    session: Session = Depends(get_session),
    principal: ApiPrincipal = Depends(get_api_principal),
) -> MailAddressLookupResponse:
    _require_scope(principal, "mail:profile:use")
    capability = _registry_capability(CAPABILITY_ADDRESSES_LOOKUP)
    if capability is None or not hasattr(capability, "lookup"):
        return MailAddressLookupResponse(available=False, candidates=[])
    candidates = getattr(capability, "lookup")(session, principal, query=query, limit=limit)
    return MailAddressLookupResponse(
        available=True,
        candidates=[MailAddressLookupCandidate.model_validate(_capability_payload(candidate)) for candidate in candidates],
    )


@router.get("/address-write-targets", response_model=MailAddressWriteTargetResponse)
def list_mail_address_write_targets(
    session: Session = Depends(get_session),
    principal: ApiPrincipal = Depends(get_api_principal),
) -> MailAddressWriteTargetResponse:
    _require_scope(principal, "mail:profile:use")
    capability = _registry_capability(CAPABILITY_ADDRESSES_CONTACT_WRITER)
    if capability is None or not hasattr(capability, "list_write_targets"):
        return MailAddressWriteTargetResponse(available=False, targets=[])
    targets = getattr(capability, "list_write_targets")(session, principal, operation="create_contact")
    return MailAddressWriteTargetResponse(
        available=True,
        targets=[MailAddressWriteTarget.model_validate(_capability_payload(target)) for target in targets],
    )


@router.post(
    "/address-contacts",
    response_model=MailContactCreateResponse,
    status_code=status.HTTP_201_CREATED,
)
def create_mail_address_contact(
    payload: MailContactCreateRequest,
    session: Session = Depends(get_session),
    principal: ApiPrincipal = Depends(get_api_principal),
) -> MailContactCreateResponse:
    _require_scope(principal, "mail:profile:use")
    capability = _registry_capability(CAPABILITY_ADDRESSES_CONTACT_WRITER)
    if capability is None or not all(
        hasattr(capability, method)
        for method in ("can_write_to_address_book", "create_contact")
    ):
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="Address-book contact writing is not available.",
        )
    decision = getattr(capability, "can_write_to_address_book")(
        session,
        principal,
        address_book_id=payload.address_book_id,
        operation="create_contact",
    )
    decision_payload = _capability_payload(decision)
    if decision_payload.get("allowed") is not True:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail=str(decision_payload.get("message") or "The selected address book is not writable."),
        )
    try:
        result = getattr(capability, "create_contact")(
            session,
            principal,
            address_book_id=payload.address_book_id,
            payload={
                "display_name": payload.display_name or payload.email,
                "emails": [{"label": "Mail", "email": payload.email, "is_primary": True}],
            },
            provenance={
                "consumer_module": MAIL_MODULE_ID,
                "consumer_workflow": "mailbox_add_contact",
            },
        )
        session.commit()
    except ValueError as exc:
        session.rollback()
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail=str(exc),
        ) from exc
    return MailContactCreateResponse.model_validate(_capability_payload(result))


@router.get("/settings/delta", response_model=MailSettingsDeltaResponse)
def mail_settings_delta(
    scope_type: str = Query(default="tenant"),
    scope_id: str | None = Query(default=None),
    include_inactive: bool = False,
    campaign_id: str | None = Query(default=None),
    since: str | None = None,
    limit: int = Query(default=100, ge=1, le=500),
    principal: ApiPrincipal = Depends(get_api_principal),
    session: Session = Depends(get_session),
):
    clean_scope_type = scope_type.strip().casefold()
    _require_any_scope(principal, "mail:profile:read", "system:settings:read")
    _require_policy_read_scope(principal, clean_scope_type)
    _require_policy_scope_visibility(
        session,
        principal=principal,
        scope_type=clean_scope_type,
        scope_id=scope_id,
    )
    _require_policy_campaign_context(
        session,
        principal=principal,
        scope_type=clean_scope_type,
        scope_id=scope_id,
        campaign_id=campaign_id,
    )
    if since is None:
        return _full_mail_settings_delta_response(
            session,
            principal=principal,
            scope_type=clean_scope_type,
            scope_id=scope_id,
            include_inactive=include_inactive,
            campaign_id=campaign_id,
        )
    entries, has_more = _mail_settings_entries(session, tenant_id=principal.tenant_id, since=since, limit=limit)
    if entries is None:
        return _full_mail_settings_delta_response(
            session,
            principal=principal,
            scope_type=clean_scope_type,
            scope_id=scope_id,
            include_inactive=include_inactive,
            campaign_id=campaign_id,
        )
    visible_profile_entries = [
        entry
        for entry in entries
        if entry.collection == MAIL_PROFILES_COLLECTION
        and entry.resource_type == MAIL_PROFILE_RESOURCE
        and _profile_change_visible_to_principal(session, entry=entry, principal=principal)
    ]
    changed_profile_ids = {
        entry.resource_id
        for entry in visible_profile_entries
    }
    visible_profiles = {
        profile.id: profile
        for profile in list_mail_server_profiles(
            session,
            tenant_id=principal.tenant_id,
            include_inactive=include_inactive,
            campaign_id=campaign_id,
            **_profile_actor_kwargs(principal, administrative_visibility=True),
        )
        if profile.id in changed_profile_ids
    }
    policy_changed = any(entry.collection == MAIL_POLICIES_COLLECTION for entry in entries)
    changed_sections = []
    if changed_profile_ids:
        changed_sections.append("profiles")
    if policy_changed:
        changed_sections.append("policy")
    deleted = [
        _mail_deleted_item(entry)
        for entry in visible_profile_entries
        if entry.resource_id not in visible_profiles
    ]
    return MailSettingsDeltaResponse(
        profiles=[
            _profile_response(
                profile,
                session=session,
                principal=principal,
                target_scope_type=clean_scope_type,
                target_scope_id=scope_id,
                include_inactive=include_inactive,
                administrative=True,
            )
            for profile in visible_profiles.values()
        ],
        policy=_policy_response(session, tenant_id=principal.tenant_id, scope_type=clean_scope_type, scope_id=scope_id, campaign_id=campaign_id) if policy_changed else None,
        changed_sections=changed_sections,
        deleted=deleted,
        watermark=_mail_settings_response_watermark(session, tenant_id=principal.tenant_id, entries=entries, has_more=has_more),
        has_more=has_more,
        full=False,
    )


@router.get("/profiles", response_model=MailServerProfileListResponse)
def list_profiles(
    include_inactive: bool = False,
    campaign_id: str | None = Query(default=None),
    principal: ApiPrincipal = Depends(get_api_principal),
    session: Session = Depends(get_session),
):
    _require_any_scope(principal, "mail:profile:read", "system:settings:read")
    try:
        profiles = list_mail_server_profiles(
            session,
            tenant_id=principal.tenant_id,
            include_inactive=include_inactive,
            campaign_id=campaign_id,
            **_profile_actor_kwargs(principal, administrative_visibility=True),
        )
        return MailServerProfileListResponse(
            profiles=[
                _profile_response(
                    profile,
                    session=session,
                    principal=principal,
                    target_scope_type="campaign" if campaign_id else None,
                    target_scope_id=campaign_id,
                    include_inactive=include_inactive,
                    administrative=include_inactive,
                )
                for profile in profiles
            ]
        )
    except MailProfileError as exc:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc


@router.post(
    "/profiles/{profile_id}/servers",
    response_model=MailServerEndpointResponse,
    status_code=status.HTTP_201_CREATED,
)
def create_profile_server(
    profile_id: str,
    payload: MailServerEndpointCreateRequest,
    principal: ApiPrincipal = Depends(get_api_principal),
    session: Session = Depends(get_session),
):
    if payload.protocol == "pop3":
        _require_scope(principal, "mail:pop3:manage")
    try:
        profile = _profile_for_mutation(
            session,
            principal=principal,
            profile_id=profile_id,
        )
        if payload.protocol == "jmap":
            _assert_jmap_policy_allows_config(
                session,
                principal=principal,
                profile=profile,
                config=payload.config,
            )
        server = create_mail_server_endpoint(
            session,
            profile=profile,
            protocol=payload.protocol,
            name=payload.name,
            config=payload.config,
            user_id=principal.user.id,
            inherit_to_lower_scopes=payload.inherit_to_lower_scopes,
            is_default=payload.is_default,
            is_active=payload.is_active,
        )
        _record_profile_child_change(
            session,
            principal=principal,
            profile=profile,
            operation="created",
            resource_type=MAIL_SERVER_RESOURCE,
            resource_id=server.id,
        )
        session.commit()
        session.refresh(server)
        return _server_response(
            session,
            principal=principal,
            profile=profile,
            server=server,
        )
    except (MailProfileError, MailServerHierarchyError) as exc:
        session.rollback()
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail=str(exc),
        ) from exc


@router.patch(
    "/profiles/{profile_id}/servers/{server_id}",
    response_model=MailServerEndpointResponse,
)
def update_profile_server(
    profile_id: str,
    server_id: str,
    payload: MailServerEndpointUpdateRequest,
    principal: ApiPrincipal = Depends(get_api_principal),
    session: Session = Depends(get_session),
):
    try:
        profile, server = _server_for_mutation(
            session,
            principal=principal,
            profile_id=profile_id,
            server_id=server_id,
        )
        if server.protocol == "pop3":
            _require_scope(principal, "mail:pop3:manage")
        if server.protocol == "jmap" and payload.config is not None:
            _assert_jmap_policy_allows_config(
                session,
                principal=principal,
                profile=profile,
                config=payload.config,
            )
        update_mail_server_endpoint(
            session,
            server=server,
            user_id=principal.user.id,
            name=payload.name,
            config=payload.config,
            inherit_to_lower_scopes=payload.inherit_to_lower_scopes,
            is_default=payload.is_default,
            is_active=payload.is_active,
        )
        _record_profile_child_change(
            session,
            principal=principal,
            profile=profile,
            operation="updated",
            resource_type=MAIL_SERVER_RESOURCE,
            resource_id=server.id,
        )
        session.commit()
        session.refresh(server)
        return _server_response(
            session,
            principal=principal,
            profile=profile,
            server=server,
        )
    except (MailProfileError, MailServerHierarchyError) as exc:
        session.rollback()
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail=str(exc),
        ) from exc


@router.delete(
    "/profiles/{profile_id}/servers/{server_id}",
    response_model=MailServerEndpointResponse,
)
def deactivate_profile_server(
    profile_id: str,
    server_id: str,
    principal: ApiPrincipal = Depends(get_api_principal),
    session: Session = Depends(get_session),
):
    try:
        profile, server = _server_for_mutation(
            session,
            principal=principal,
            profile_id=profile_id,
            server_id=server_id,
        )
        if server.protocol == "pop3":
            _require_scope(principal, "mail:pop3:manage")
        update_mail_server_endpoint(
            session,
            server=server,
            user_id=principal.user.id,
            is_active=False,
        )
        _record_profile_child_change(
            session,
            principal=principal,
            profile=profile,
            operation="deactivated",
            resource_type=MAIL_SERVER_RESOURCE,
            resource_id=server.id,
        )
        session.commit()
        session.refresh(server)
        return _server_response(
            session,
            principal=principal,
            profile=profile,
            server=server,
        )
    except MailServerHierarchyError as exc:
        session.rollback()
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail=str(exc),
        ) from exc


def _credential_hierarchy_context(principal: ApiPrincipal, profile) -> MailHierarchyContext:
    can_manage_all = has_scope(principal, "mail:secret:manage") or (
        profile.scope_type == "system"
        and has_scope(principal, "system:settings:write")
    )
    return hierarchy_context_for_profile(
        profile,
        user_id=principal.user.id,
        group_ids=principal.group_ids,
        administrative=can_manage_all,
    )


@router.get(
    "/profiles/{profile_id}/servers/{server_id}/available-credentials",
    response_model=MailCredentialListResponse,
)
def list_profile_server_available_credentials(
    profile_id: str,
    server_id: str,
    include_inactive: bool = False,
    principal: ApiPrincipal = Depends(get_api_principal),
    session: Session = Depends(get_session),
):
    profile = _get_profile_for_principal(
        session,
        principal=principal,
        profile_id=profile_id,
        administrative_visibility=True,
    )
    _require_profile_credentials_scope(
        principal,
        profile.scope_type or "tenant",
        profile.scope_id,
    )
    context = _credential_hierarchy_context(principal, profile)
    try:
        server = get_mail_server_endpoint(
            session,
            profile=profile,
            server_id=server_id,
            context=context,
            require_active=False,
        )
        credentials = list_available_mail_credentials(
            session,
            context=context,
            server_id=server.id,
            include_inactive=include_inactive,
        )
        return MailCredentialListResponse(
            credentials=[
                MailCredentialEnvelopeResponse.model_validate(
                    mail_credential_payload(credential)
                )
                for credential in credentials
            ]
        )
    except MailServerHierarchyError as exc:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=str(exc),
        ) from exc


@router.post(
    "/profiles/{profile_id}/servers/{server_id}/credentials",
    response_model=MailCredentialEnvelopeResponse,
    status_code=status.HTTP_201_CREATED,
)
def create_profile_server_credential(
    profile_id: str,
    server_id: str,
    payload: MailCredentialCreateRequest,
    principal: ApiPrincipal = Depends(get_api_principal),
    session: Session = Depends(get_session),
):
    try:
        profile, server = _server_for_mutation(
            session,
            principal=principal,
            profile_id=profile_id,
            server_id=server_id,
        )
        _require_profile_credentials_scope(
            principal,
            profile.scope_type or "tenant",
            profile.scope_id,
        )
        credential = create_and_bind_mail_credential(
            session,
            profile=profile,
            server=server,
            name=payload.name,
            description=payload.description,
            credential_kind=payload.credential_kind,
            username=payload.username,
            password=payload.password,
            public_data=payload.public_data,
            secret_data=payload.secret_data,
            user_id=principal.user.id,
            inherit_to_lower_scopes=payload.inherit_to_lower_scopes,
            allowed_modules=payload.allowed_modules,
            allowed_server_refs=payload.allowed_server_refs,
            is_default=payload.is_default,
        )
        binding, _ = get_bound_mail_credential(
            session,
            server=server,
            credential_id=credential.id,
            context=_credential_hierarchy_context(principal, profile),
            require_active=False,
        )
        _record_profile_child_change(
            session,
            principal=principal,
            profile=profile,
            operation="created",
            resource_type=MAIL_CREDENTIAL_RESOURCE,
            resource_id=credential.id,
        )
        session.commit()
        session.refresh(credential)
        return MailCredentialEnvelopeResponse.model_validate(
            mail_credential_payload(
                credential,
                server_id=server.id,
                binding_id=binding.id,
                is_default=binding.is_default,
            )
        )
    except (MailProfileError, MailServerHierarchyError) as exc:
        session.rollback()
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail=str(exc),
        ) from exc


@router.post(
    "/profiles/{profile_id}/campaign-credentials",
    response_model=MailCredentialEnvelopeResponse,
    status_code=status.HTTP_201_CREATED,
)
def create_campaign_profile_credential(
    profile_id: str,
    payload: MailCampaignCredentialCreateRequest,
    campaign_id: str = Query(...),
    principal: ApiPrincipal = Depends(get_api_principal),
    session: Session = Depends(get_session),
):
    _require_any_scope(
        principal,
        "campaigns:campaign:update",
        "mail:secret:manage",
    )
    _require_campaign_context_visibility(
        session,
        principal=principal,
        campaign_id=campaign_id,
    )
    try:
        profile = next(
            (
                item
                for item in list_mail_server_profiles(
                    session,
                    tenant_id=principal.tenant_id,
                    campaign_id=campaign_id,
                    **_profile_actor_kwargs(
                        principal,
                        administrative_visibility=False,
                    ),
                )
                if item.id == profile_id
            ),
            None,
        )
        if profile is None:
            raise MailServerHierarchyError("Mail-server profile not found")
        context = hierarchy_context_for_profile(
            profile,
            user_id=principal.user.id,
            group_ids=principal.group_ids,
            target_scope_type="campaign",
            target_scope_id=campaign_id,
        )
        server_ids = list(dict.fromkeys(item.strip() for item in payload.server_ids if item.strip()))
        servers = [
            get_mail_server_endpoint(
                session,
                profile=profile,
                server_id=server_id,
                context=context,
                require_active=True,
                for_update=True,
            )
            for server_id in server_ids
        ]
        if not servers:
            raise MailServerHierarchyError("Select at least one mail server")
        allowed_server_refs = [
            ref
            for ref in (mail_server_ref(server.id) for server in servers)
            if ref is not None
        ]
        credential = create_and_bind_mail_credential(
            session,
            profile=profile,
            server=servers[0],
            name=payload.name,
            username=payload.username,
            password=payload.password.get_secret_value(),
            user_id=principal.user.id,
            inherit_to_lower_scopes=False,
            allowed_modules=("mail",),
            allowed_server_refs=allowed_server_refs,
            credential_scope_type="campaign",
            credential_scope_id=campaign_id,
        )
        for server in servers[1:]:
            bind_mail_credential(
                session,
                server=server,
                credential=credential,
                user_id=principal.user.id,
            )
        first_binding, _ = get_bound_mail_credential(
            session,
            server=servers[0],
            credential_id=credential.id,
            context=context,
        )
        _record_profile_child_change(
            session,
            principal=principal,
            profile=profile,
            operation="created",
            resource_type=MAIL_CREDENTIAL_RESOURCE,
            resource_id=credential.id,
        )
        session.commit()
        session.refresh(credential)
        return MailCredentialEnvelopeResponse.model_validate(
            mail_credential_payload(
                credential,
                server_id=servers[0].id,
                binding_id=first_binding.id,
                is_default=False,
            )
        )
    except (MailProfileError, MailServerHierarchyError) as exc:
        session.rollback()
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail=str(exc),
        ) from exc


@router.post(
    "/profiles/{profile_id}/servers/{server_id}/credential-bindings",
    response_model=MailCredentialEnvelopeResponse,
    status_code=status.HTTP_201_CREATED,
)
def bind_profile_server_credential(
    profile_id: str,
    server_id: str,
    payload: MailCredentialBindRequest,
    principal: ApiPrincipal = Depends(get_api_principal),
    session: Session = Depends(get_session),
):
    try:
        profile, server = _server_for_mutation(
            session,
            principal=principal,
            profile_id=profile_id,
            server_id=server_id,
        )
        _require_profile_credentials_scope(
            principal,
            profile.scope_type or "tenant",
            profile.scope_id,
        )
        context = _credential_hierarchy_context(principal, profile)
        credential = get_available_mail_credential(
            session,
            credential_id=payload.credential_id,
            context=context,
            server_id=server.id,
        )
        binding = bind_mail_credential(
            session,
            server=server,
            credential=credential,
            user_id=principal.user.id,
            is_default=payload.is_default,
        )
        _record_profile_child_change(
            session,
            principal=principal,
            profile=profile,
            operation="bound",
            resource_type=MAIL_CREDENTIAL_RESOURCE,
            resource_id=credential.id,
        )
        session.commit()
        session.refresh(credential)
        return MailCredentialEnvelopeResponse.model_validate(
            mail_credential_payload(
                credential,
                server_id=server.id,
                binding_id=binding.id,
                is_default=binding.is_default,
            )
        )
    except (MailProfileError, MailServerHierarchyError) as exc:
        session.rollback()
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail=str(exc),
        ) from exc


@router.patch(
    "/profiles/{profile_id}/servers/{server_id}/credentials/{credential_id}",
    response_model=MailCredentialEnvelopeResponse,
)
def update_profile_server_credential(
    profile_id: str,
    server_id: str,
    credential_id: str,
    payload: MailCredentialUpdateRequest,
    principal: ApiPrincipal = Depends(get_api_principal),
    session: Session = Depends(get_session),
):
    try:
        profile, server = _server_for_mutation(
            session,
            principal=principal,
            profile_id=profile_id,
            server_id=server_id,
        )
        _require_profile_credentials_scope(
            principal,
            profile.scope_type or "tenant",
            profile.scope_id,
        )
        context = _credential_hierarchy_context(principal, profile)
        binding, credential = get_bound_mail_credential(
            session,
            server=server,
            credential_id=credential_id,
            context=context,
            require_active=False,
            for_update=True,
        )
        update_bound_mail_credential(
            session,
            server=server,
            credential=credential,
            context=context,
            user_id=principal.user.id,
            name=payload.name,
            description=payload.description,
            description_supplied="description" in payload.model_fields_set,
            username=payload.username,
            username_supplied="username" in payload.model_fields_set,
            password=payload.password,
            password_supplied="password" in payload.model_fields_set,
            public_data=payload.public_data,
            secret_data=payload.secret_data,
            allowed_modules=payload.allowed_modules,
            allowed_server_refs=payload.allowed_server_refs,
            inherit_to_lower_scopes=payload.inherit_to_lower_scopes,
            is_active=payload.is_active,
            is_default=payload.is_default,
        )
        binding, credential = get_bound_mail_credential(
            session,
            server=server,
            credential_id=credential_id,
            context=context,
            require_active=False,
        )
        _record_profile_child_change(
            session,
            principal=principal,
            profile=profile,
            operation="updated",
            resource_type=MAIL_CREDENTIAL_RESOURCE,
            resource_id=credential.id,
        )
        session.commit()
        session.refresh(credential)
        return MailCredentialEnvelopeResponse.model_validate(
            mail_credential_payload(
                credential,
                server_id=server.id,
                binding_id=binding.id,
                is_default=binding.is_default,
            )
        )
    except (MailProfileError, MailServerHierarchyError) as exc:
        session.rollback()
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail=str(exc),
        ) from exc


@router.delete(
    "/profiles/{profile_id}/servers/{server_id}/credentials/{credential_id}",
    status_code=status.HTTP_204_NO_CONTENT,
)
def unlink_profile_server_credential(
    profile_id: str,
    server_id: str,
    credential_id: str,
    retire_if_unused: bool = Query(default=False),
    principal: ApiPrincipal = Depends(get_api_principal),
    session: Session = Depends(get_session),
):
    try:
        profile, server = _server_for_mutation(
            session,
            principal=principal,
            profile_id=profile_id,
            server_id=server_id,
        )
        _require_profile_credentials_scope(
            principal,
            profile.scope_type or "tenant",
            profile.scope_id,
        )
        context = _credential_hierarchy_context(principal, profile)
        _, credential = get_bound_mail_credential(
            session,
            server=server,
            credential_id=credential_id,
            context=context,
            require_active=False,
        )
        unlink_mail_credential(
            session,
            server=server,
            credential=credential,
            context=context,
            user_id=principal.user.id,
            retire_if_unused=retire_if_unused,
        )
        _record_profile_child_change(
            session,
            principal=principal,
            profile=profile,
            operation="unlinked",
            resource_type=MAIL_CREDENTIAL_RESOURCE,
            resource_id=credential.id,
        )
        session.commit()
        return None
    except (MailProfileError, MailServerHierarchyError) as exc:
        session.rollback()
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail=str(exc),
        ) from exc


@router.post("/profiles", response_model=MailServerProfileResponse, status_code=status.HTTP_201_CREATED)
def create_profile(
    payload: MailServerProfileCreateRequest,
    principal: ApiPrincipal = Depends(get_api_principal),
    session: Session = Depends(get_session),
):
    authorization_scope_id = (
        payload.scope_id or principal.user.id
        if payload.scope_type == "user"
        else payload.scope_id
    )
    _require_profile_write_scope(principal, payload.scope_type, authorization_scope_id)
    _require_campaign_profile_mutation_access(
        session,
        principal=principal,
        scope_type=payload.scope_type,
        scope_id=payload.scope_id,
    )
    if payload.credentials_supplied():
        _require_profile_credentials_scope(
            principal,
            payload.scope_type,
            authorization_scope_id,
    )
    try:
        smtp_config = payload.smtp_config()
        imap_config = payload.imap_config()
        profile = create_mail_server_profile(
            session,
            tenant_id=principal.tenant_id,
            user_id=principal.user.id,
            name=payload.name,
            slug=payload.slug,
            description=payload.description,
            smtp=smtp_config,
            imap=imap_config,
            is_active=payload.is_active,
            inherit_to_lower_scopes=payload.inherit_to_lower_scopes,
            scope_type=payload.scope_type,
            scope_id=payload.scope_id,
        )
        initialize_profile_hierarchy(
            session,
            profile=profile,
            smtp=smtp_config,
            imap=imap_config,
            user_id=principal.user.id,
        )
        _record_mail_change(
            session,
            collection=MAIL_PROFILES_COLLECTION,
            resource_type=MAIL_PROFILE_RESOURCE,
            resource_id=profile.id,
            operation="created",
            principal=principal,
            tenant_id=profile.tenant_id,
            payload={"scope_type": profile.scope_type, "scope_id": profile.scope_id},
        )
        session.commit()
        session.refresh(profile)
        return _profile_response(
            profile,
            session=session,
            principal=principal,
            target_scope_type=profile.scope_type,
            target_scope_id=profile.scope_id,
            include_inactive=True,
            administrative=True,
        )
    except (MailProfileError, MailServerHierarchyError) as exc:
        session.rollback()
        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc


@router.get("/profiles/{profile_id}", response_model=MailServerProfileResponse)
def get_profile(
    profile_id: str,
    principal: ApiPrincipal = Depends(get_api_principal),
    session: Session = Depends(get_session),
):
    _require_any_scope(principal, "mail:profile:read", "system:settings:read")
    try:
        profile = _get_profile_for_principal(
                session,
                principal=principal,
                profile_id=profile_id,
                administrative_visibility=True,
            )
        return _profile_response(
            profile,
            session=session,
            principal=principal,
            target_scope_type=profile.scope_type,
            target_scope_id=profile.scope_id,
            include_inactive=True,
            administrative=True,
        )
    except MailProfileError as exc:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc


@router.patch("/profiles/{profile_id}", response_model=MailServerProfileResponse)
def update_profile(
    profile_id: str,
    payload: MailServerProfileUpdateRequest,
    principal: ApiPrincipal = Depends(get_api_principal),
    session: Session = Depends(get_session),
):
    try:
        profile = _profile_for_mutation(
            session,
            principal=principal,
            profile_id=profile_id,
        )
        _require_campaign_profile_mutation_access(
            session,
            principal=principal,
            scope_type=profile.scope_type or "tenant",
            scope_id=profile.scope_id,
        )
        smtp_config = payload.smtp_config()
        if payload.smtp is None and "smtp" in payload.credentials.model_fields_set:
            smtp_data = dict(profile.smtp_config or {})
            smtp_data.update(payload.credentials.smtp.model_dump(mode="json", exclude_unset=True))
            smtp_config = SmtpConfig.model_validate(smtp_data)
        imap_config = payload.imap_config()
        if payload.imap is None and "imap" in payload.credentials.model_fields_set and profile.imap_config:
            imap_data = dict(profile.imap_config or {})
            imap_data.update(payload.credentials.imap.model_dump(mode="json", exclude_unset=True))
            imap_config = ImapConfig.model_validate(imap_data)
        if _profile_update_requires_credentials_scope(
            profile,
            payload=payload,
            smtp=smtp_config,
            imap=imap_config,
        ):
            _require_profile_credentials_scope(
                principal,
                profile.scope_type or "tenant",
                profile.scope_id,
            )
        update_mail_server_profile(
            session,
            profile,
            tenant_id=principal.tenant_id,
            user_id=principal.user.id,
            api_key_id=principal.api_key_id,
            name=payload.name,
            slug=payload.slug,
            description=payload.description if "description" in payload.model_fields_set else None,
            is_active=payload.is_active,
            inherit_to_lower_scopes=payload.inherit_to_lower_scopes,
            smtp=smtp_config,
            imap=imap_config,
            clear_imap=payload.clear_imap,
        )
        smtp_server = None
        if smtp_config is not None:
            smtp_server = sync_default_profile_server(
                session,
                profile=profile,
                protocol="smtp",
                config=smtp_config.model_dump(mode="json", exclude_none=True),
                user_id=principal.user.id,
            )
        imap_server = None
        if imap_config is not None or payload.clear_imap:
            imap_server = sync_default_profile_server(
                session,
                profile=profile,
                protocol="imap",
                config=(
                    imap_config.model_dump(mode="json", exclude_none=True)
                    if imap_config is not None and not payload.clear_imap
                    else None
                ),
                user_id=principal.user.id,
            )
        if smtp_server is not None and "smtp" in payload.credentials.model_fields_set:
            smtp_credentials = payload.credentials.smtp
            sync_default_profile_credential(
                session,
                profile=profile,
                server=smtp_server,
                username=smtp_credentials.username,
                username_supplied="username" in smtp_credentials.model_fields_set,
                password=smtp_credentials.password,
                password_supplied="password" in smtp_credentials.model_fields_set,
                user_id=principal.user.id,
            )
        if imap_server is not None and "imap" in payload.credentials.model_fields_set:
            imap_credentials = payload.credentials.imap
            sync_default_profile_credential(
                session,
                profile=profile,
                server=imap_server,
                username=imap_credentials.username,
                username_supplied="username" in imap_credentials.model_fields_set,
                password=imap_credentials.password,
                password_supplied="password" in imap_credentials.model_fields_set,
                user_id=principal.user.id,
            )
        _record_mail_change(
            session,
            collection=MAIL_PROFILES_COLLECTION,
            resource_type=MAIL_PROFILE_RESOURCE,
            resource_id=profile.id,
            operation="updated",
            principal=principal,
            tenant_id=profile.tenant_id,
            payload={"scope_type": profile.scope_type, "scope_id": profile.scope_id},
        )
        session.commit()
        session.refresh(profile)
        return _profile_response(
            profile,
            session=session,
            principal=principal,
            target_scope_type=profile.scope_type,
            target_scope_id=profile.scope_id,
            include_inactive=True,
            administrative=True,
        )
    except (MailProfileError, MailServerHierarchyError) as exc:
        session.rollback()
        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc
    except Exception:
        session.rollback()
        raise


@router.delete("/profiles/{profile_id}", response_model=MailServerProfileResponse)
def deactivate_profile(
    profile_id: str,
    principal: ApiPrincipal = Depends(get_api_principal),
    session: Session = Depends(get_session),
):
    try:
        profile = _profile_for_mutation(
            session,
            principal=principal,
            profile_id=profile_id,
        )
        _require_campaign_profile_mutation_access(
            session,
            principal=principal,
            scope_type=profile.scope_type or "tenant",
            scope_id=profile.scope_id,
        )
        if _profile_has_stored_credentials(profile):
            _require_profile_credentials_scope(
                principal,
                profile.scope_type or "tenant",
                profile.scope_id,
            )
        was_active = bool(profile.is_active)
        profile.is_active = False
        deleted_protocols = delete_mail_profile_credentials(
            session,
            profile=profile,
            deletion_reason="profile_deactivated",
            user_id=principal.user.id,
            api_key_id=principal.api_key_id,
        )
        # Deactivation invalidates unauthenticated IMAP profiles too. Keep the
        # cache deletion in this transaction so a failed audit/change record
        # cannot leave profile state and cached mailbox data out of sync.
        clear_mailbox_index(session, profile_id=profile.id)
        session.add(profile)
        if was_active or deleted_protocols:
            _record_mail_change(
                session,
                collection=MAIL_PROFILES_COLLECTION,
                resource_type=MAIL_PROFILE_RESOURCE,
                resource_id=profile.id,
                operation="deleted",
                principal=principal,
                tenant_id=profile.tenant_id,
                payload={
                    "scope_type": profile.scope_type,
                    "scope_id": profile.scope_id,
                    "credentials_deleted": bool(deleted_protocols),
                    "deleted_credential_protocols": list(deleted_protocols),
                },
            )
        session.commit()
        session.refresh(profile)
        return _profile_response(
            profile,
            session=session,
            principal=principal,
            target_scope_type=profile.scope_type,
            target_scope_id=profile.scope_id,
            include_inactive=True,
            administrative=True,
        )
    except MailProfileError as exc:
        session.rollback()
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
    except Exception:
        session.rollback()
        raise


@router.get("/policies/{scope_type}", response_model=MailProfilePolicyResponse)
def read_mail_profile_policy(
    scope_type: str,
    scope_id: str | None = Query(default=None),
    campaign_id: str | None = Query(default=None),
    principal: ApiPrincipal = Depends(get_api_principal),
    session: Session = Depends(get_session),
):
    scope_type = scope_type.strip().casefold()
    _require_policy_read_scope(principal, scope_type)
    _require_policy_scope_visibility(
        session,
        principal=principal,
        scope_type=scope_type,
        scope_id=scope_id,
    )
    _require_policy_campaign_context(
        session,
        principal=principal,
        scope_type=scope_type,
        scope_id=scope_id,
        campaign_id=campaign_id,
    )
    try:
        return _policy_response(session, tenant_id=principal.tenant_id, scope_type=scope_type, scope_id=scope_id, campaign_id=campaign_id)
    except MailProfileError as exc:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc


@router.put("/policies/{scope_type}", response_model=MailProfilePolicyResponse)
def write_mail_profile_policy(
    scope_type: str,
    payload: MailProfilePolicyUpdateRequest,
    scope_id: str | None = Query(default=None),
    principal: ApiPrincipal = Depends(get_api_principal),
    session: Session = Depends(get_session),
):
    scope_type = scope_type.strip().casefold()
    _require_policy_write_scope(principal, scope_type)
    _require_policy_scope_visibility(
        session,
        principal=principal,
        scope_type=scope_type,
        scope_id=scope_id,
    )
    _require_policy_campaign_context(
        session,
        principal=principal,
        scope_type=scope_type,
        scope_id=scope_id,
        campaign_id=None,
    )
    try:
        set_mail_profile_policy(
            session,
            tenant_id=principal.tenant_id,
            scope_type=scope_type,
            scope_id=scope_id,
            policy=payload.policy.model_dump(mode="json"),
        )
        _record_mail_change(
            session,
            collection=MAIL_POLICIES_COLLECTION,
            resource_type=MAIL_POLICY_RESOURCE,
            resource_id=f"{scope_type}:{scope_id or ''}",
            operation="updated",
            principal=principal,
            tenant_id=None if scope_type == "system" else principal.tenant_id,
            payload={"scope_type": scope_type, "scope_id": scope_id},
        )
        session.commit()
        return _policy_response(session, tenant_id=principal.tenant_id, scope_type=scope_type, scope_id=scope_id)
    except MailProfileError as exc:
        session.rollback()
        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc


@router.post("/profiles/{profile_id}/test-smtp", response_model=MailConnectionTestResponse)
def test_profile_smtp(
    profile_id: str,
    server_id: str | None = Query(default=None),
    credential_id: str | None = Query(default=None),
    campaign_id: str | None = Query(default=None),
    principal: ApiPrincipal = Depends(get_api_principal),
    session: Session = Depends(get_session),
):
    _require_scope(principal, "mail:profile:test")
    _require_scope(principal, "mail:profile:use")
    try:
        profile = _get_profile_for_principal(
            session,
            principal=principal,
            profile_id=profile_id,
            require_active=True,
        )
        if callable(getattr(session, "execute", None)):
            smtp = resolve_mail_transport(
                session,
                profile=profile,
                protocol="smtp",
                context=_transport_context_for_principal(
                    session,
                    principal=principal,
                    campaign_id=campaign_id,
                ),
                server_id=server_id,
                credential_id=credential_id,
            ).config
        else:
            smtp = smtp_config_from_profile(profile)
        result = test_smtp_login(smtp_config=smtp)
        return MailConnectionTestResponse(ok=True, protocol="smtp", host=result.host, port=result.port, security=result.security, message="SMTP connection successful.", details={"authenticated": result.authenticated})
    except (MailProfileError, MailServerHierarchyError) as exc:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
    except Exception as exc:
        return MailConnectionTestResponse(ok=False, protocol="smtp", message=_safe_error_message(exc), details={"error_type": exc.__class__.__name__})


@router.post("/profiles/{profile_id}/test-imap", response_model=MailConnectionTestResponse)
def test_profile_imap(
    profile_id: str,
    server_id: str | None = Query(default=None),
    credential_id: str | None = Query(default=None),
    campaign_id: str | None = Query(default=None),
    principal: ApiPrincipal = Depends(get_api_principal),
    session: Session = Depends(get_session),
):
    _require_scope(principal, "mail:profile:test")
    _require_scope(principal, "mail:profile:use")
    try:
        profile = _get_profile_for_principal(
            session,
            principal=principal,
            profile_id=profile_id,
            require_active=True,
        )
        if callable(getattr(session, "execute", None)):
            imap = resolve_mail_transport(
                session,
                profile=profile,
                protocol="imap",
                context=_transport_context_for_principal(
                    session,
                    principal=principal,
                    campaign_id=campaign_id,
                ),
                server_id=server_id,
                credential_id=credential_id,
            ).config
        else:
            imap = imap_config_from_profile(profile)
            if imap is None:
                raise MailProfileError("Mail-server profile has no IMAP configuration")
        result = test_imap_login(imap_config=imap)
        return MailConnectionTestResponse(ok=True, protocol="imap", host=result.host, port=result.port, security=result.security, message="IMAP connection successful.", details={"authenticated": result.authenticated})
    except (MailProfileError, MailServerHierarchyError) as exc:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
    except Exception as exc:
        return MailConnectionTestResponse(ok=False, protocol="imap", message=_safe_error_message(exc), details={"error_type": exc.__class__.__name__})


@router.post("/profiles/{profile_id}/test-jmap", response_model=MailConnectionTestResponse)
def test_profile_jmap(
    profile_id: str,
    server_id: str | None = Query(default=None),
    credential_id: str | None = Query(default=None),
    campaign_id: str | None = Query(default=None),
    principal: ApiPrincipal = Depends(get_api_principal),
    session: Session = Depends(get_session),
):
    _require_scope(principal, "mail:profile:test")
    _require_scope(principal, "mail:profile:use")
    try:
        jmap = _jmap_config_for_principal(
            session,
            principal=principal,
            profile_id=profile_id,
            server_id=server_id,
            credential_id=credential_id,
            campaign_id=campaign_id,
        )
        result = test_jmap_connection(jmap_config=jmap)
        return MailConnectionTestResponse(
            ok=True,
            protocol="jmap",
            host=result.host,
            port=result.port,
            security=result.security,
            message="JMAP connection successful.",
            details={
                "authenticated": result.authenticated,
                "account_id": result.account_id,
                "session_state": result.session_state,
                "capabilities": list(result.capabilities),
            },
        )
    except MailProfileError as exc:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
    except (JmapConfigurationError, JmapCapabilityError) as exc:
        return MailConnectionTestResponse(
            ok=False,
            protocol="jmap",
            message=str(exc),
            details={"error_type": exc.__class__.__name__},
        )
    except JmapAuthenticationError as exc:
        return MailConnectionTestResponse(
            ok=False,
            protocol="jmap",
            message=str(exc),
            details={"error_type": "authentication"},
        )
    except JmapPermissionError as exc:
        return MailConnectionTestResponse(
            ok=False,
            protocol="jmap",
            message=str(exc),
            details={"error_type": "permission"},
        )
    except JmapProviderError as exc:
        return MailConnectionTestResponse(
            ok=False,
            protocol="jmap",
            message=str(exc),
            details={"error_type": "transient_provider"},
        )


def _resolve_profile_pop3_transport(
    session: Session,
    *,
    principal: ApiPrincipal,
    profile_id: str,
    server_id: str,
    credential_id: str | None,
):
    profile = _get_profile_for_principal(
        session,
        principal=principal,
        profile_id=profile_id,
        require_active=True,
    )
    resolved = resolve_mail_transport(
        session,
        profile=profile,
        protocol="pop3",
        context=_transport_context_for_principal(
            session,
            principal=principal,
        ),
        server_id=server_id,
        credential_id=credential_id,
    )
    if resolved.server is None or not isinstance(resolved.config, Pop3Config):
        raise MailServerHierarchyError("The selected POP3 server is unavailable")
    return profile, resolved


@router.post(
    "/profiles/{profile_id}/test-pop3",
    response_model=MailConnectionTestResponse,
)
def test_profile_pop3(
    profile_id: str,
    server_id: str = Query(...),
    credential_id: str | None = Query(default=None),
    principal: ApiPrincipal = Depends(get_api_principal),
    session: Session = Depends(get_session),
):
    _require_scope(principal, "mail:profile:test")
    _require_scope(principal, "mail:profile:use")
    _require_any_scope(principal, "mail:pop3:manage", "mail:pop3:import")
    try:
        _profile, resolved = _resolve_profile_pop3_transport(
            session,
            principal=principal,
            profile_id=profile_id,
            server_id=server_id,
            credential_id=credential_id,
        )
        result = test_pop3_login(pop3_config=resolved.config)
        return MailConnectionTestResponse(
            ok=True,
            protocol="pop3",
            host=result.host,
            port=result.port,
            security=result.security,
            message="POP3 connection successful.",
            details={
                "authenticated": result.authenticated,
                "message_count": result.message_count,
                "mailbox_size_bytes": result.mailbox_size_bytes,
                "legacy_import": True,
            },
        )
    except (MailProfileError, MailServerHierarchyError) as exc:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=str(exc),
        ) from exc
    except (Pop3ConfigurationError, Pop3ProviderError) as exc:
        return MailConnectionTestResponse(
            ok=False,
            protocol="pop3",
            message=_safe_error_message(exc),
            details={"error_type": exc.__class__.__name__, "legacy_import": True},
        )


@router.post(
    "/profiles/{profile_id}/pop3/preview",
    response_model=MailPop3PreviewResponse,
)
def preview_profile_pop3_import(
    profile_id: str,
    payload: MailPop3PreviewRequest,
    principal: ApiPrincipal = Depends(get_api_principal),
    session: Session = Depends(get_session),
):
    _require_scope(principal, "mail:profile:use")
    _require_scope(principal, "mail:pop3:import")
    try:
        _profile, resolved = _resolve_profile_pop3_transport(
            session,
            principal=principal,
            profile_id=profile_id,
            server_id=payload.server_id,
            credential_id=payload.credential_id,
        )
        result = preview_pop3_messages(
            pop3_config=resolved.config,
            limit=payload.limit,
        )
        uidls = [message.uidl for message in result.messages]
        imported_uidls = set(
            session.scalars(
                select(MailPop3Import.provider_uidl).where(
                    MailPop3Import.tenant_id == principal.tenant_id,
                    MailPop3Import.profile_id == profile_id,
                    MailPop3Import.pop3_server_id == resolved.server.id,
                    MailPop3Import.provider_uidl.in_(uidls),
                )
            )
        ) if uidls else set()
        return MailPop3PreviewResponse(
            profile_id=profile_id,
            server_id=resolved.server.id,
            transport_revision=resolved.transport_revision,
            host=result.host,
            port=result.port,
            security=result.security,
            message_count=result.message_count,
            mailbox_size_bytes=result.mailbox_size_bytes,
            delete_after_import_allowed=resolved.config.allow_delete_after_import,
            messages=[
                MailPop3MessagePreviewResponse(
                    message_number=message.message_number,
                    uidl=message.uidl,
                    subject=message.subject,
                    from_header=message.from_header,
                    to_header=message.to_header,
                    date=message.date,
                    message_id=message.message_id,
                    size_bytes=message.size_bytes,
                    body_preview=message.body_preview,
                    already_imported=message.uidl in imported_uidls,
                )
                for message in result.messages
            ],
        )
    except (MailProfileError, MailServerHierarchyError) as exc:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=str(exc),
        ) from exc
    except Pop3ConfigurationError as exc:
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail=str(exc),
        ) from exc
    except Pop3ProviderError as exc:
        raise HTTPException(
            status_code=status.HTTP_502_BAD_GATEWAY,
            detail=str(exc),
        ) from exc


@router.post(
    "/profiles/{profile_id}/pop3/import",
    response_model=MailPop3ImportResponse,
)
def import_profile_pop3_messages(
    profile_id: str,
    payload: MailPop3ImportRequest,
    principal: ApiPrincipal = Depends(get_api_principal),
    session: Session = Depends(get_session),
):
    _require_scope(principal, "mail:profile:use")
    _require_scope(principal, "mail:pop3:import")
    if payload.delete_after_import:
        _require_scope(principal, "mail:pop3:delete")
    try:
        _profile, resolved = _resolve_profile_pop3_transport(
            session,
            principal=principal,
            profile_id=profile_id,
            server_id=payload.server_id,
            credential_id=payload.credential_id,
        )
        if resolved.transport_revision != payload.expected_transport_revision:
            raise HTTPException(
                status_code=status.HTTP_409_CONFLICT,
                detail="The POP3 server or credential selection changed; refresh the preview before importing",
            )
        if payload.delete_after_import and not resolved.config.allow_delete_after_import:
            raise HTTPException(
                status_code=status.HTTP_409_CONFLICT,
                detail="Delete-after-import is disabled for the selected POP3 server",
            )
        downloaded = download_pop3_messages(
            pop3_config=resolved.config,
            uidls=payload.uidls,
        )
        imported = create_pop3_imports(
            session,
            tenant_id=principal.tenant_id,
            profile_id=profile_id,
            pop3_server_id=resolved.server.id,
            pop3_credential_id=(
                resolved.credential.id if resolved.credential is not None else None
            ),
            transport_revision=resolved.transport_revision,
            messages=downloaded,
            user_id=principal.user.id,
            deletion_requested=payload.delete_after_import,
        )
        aggregate_digest = hashlib.sha256(
            "|".join(sorted(item.raw_sha256 for item in downloaded)).encode("ascii")
        ).hexdigest()
        audit_event(
            session,
            tenant_id=principal.tenant_id,
            user_id=principal.user.id,
            action="mail.pop3.imported",
            object_type="mail_pop3_import_batch",
            object_id=aggregate_digest,
            details={
                "profile_id": profile_id,
                "server_id": resolved.server.id,
                "transport_revision": resolved.transport_revision,
                "selected_count": len(downloaded),
                "imported_count": len(imported.imported),
                "duplicate_count": len(imported.duplicates),
                "delete_after_import": payload.delete_after_import,
                "content_digest": aggregate_digest,
            },
        )
        # The governed local copy and its audit evidence become durable before
        # any separately authorized provider deletion is attempted.
        session.commit()

        deletion_status = "not_requested"
        if payload.delete_after_import:
            if not imported.imported:
                deletion_status = "skipped_no_new_messages"
            else:
                new_uidls = [row.provider_uidl for row in imported.imported]
                try:
                    delete_pop3_messages(
                        pop3_config=resolved.config,
                        uidls=new_uidls,
                    )
                    deletion_status = "succeeded"
                    deletion_error = None
                except Pop3ProviderError as exc:
                    deletion_status = (
                        "outcome_unknown" if exc.outcome_unknown else "failed"
                    )
                    deletion_error = str(exc)
                rows = mark_pop3_deletion_result(
                    session,
                    tenant_id=principal.tenant_id,
                    import_ids=[row.id for row in imported.imported],
                    status=deletion_status,
                    error=deletion_error,
                )
                # Persist the provider outcome independently of the audit
                # projection. If audit insertion is unavailable after the
                # irreversible provider operation, the import record still
                # retains the result for reconciliation and the pre-effect
                # import audit already proves that deletion was requested.
                session.commit()
                audit_event(
                    session,
                    tenant_id=principal.tenant_id,
                    user_id=principal.user.id,
                    action="mail.pop3.source_deletion",
                    object_type="mail_pop3_import_batch",
                    object_id=aggregate_digest,
                    details={
                        "profile_id": profile_id,
                        "server_id": resolved.server.id,
                        "import_count": len(rows),
                        "status": deletion_status,
                    },
                )
                session.commit()

        return MailPop3ImportResponse(
            imports=[
                MailPop3ImportRecordResponse.model_validate(
                    pop3_import_payload(row)
                )
                for row in imported.imported
            ],
            duplicate_uidls=sorted(row.provider_uidl for row in imported.duplicates),
            deletion_status=deletion_status,
        )
    except HTTPException:
        session.rollback()
        raise
    except (MailProfileError, MailServerHierarchyError) as exc:
        session.rollback()
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=str(exc),
        ) from exc
    except (Pop3ConfigurationError, Pop3ImportError) as exc:
        session.rollback()
        raise HTTPException(
            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
            detail=str(exc),
        ) from exc
    except Pop3ProviderError as exc:
        session.rollback()
        raise HTTPException(
            status_code=status.HTTP_502_BAD_GATEWAY,
            detail=str(exc),
        ) from exc
    except Exception:
        session.rollback()
        raise


@router.get("/pop3/imports", response_model=MailPop3ImportListResponse)
def get_pop3_imports(
    profile_id: str | None = Query(default=None),
    limit: int = Query(default=100, ge=1, le=500),
    principal: ApiPrincipal = Depends(get_api_principal),
    session: Session = Depends(get_session),
):
    _require_scope(principal, "mail:pop3:import")
    visible_profile_ids = {
        profile.id
        for profile in list_mail_server_profiles(
            session,
            tenant_id=principal.tenant_id,
            include_inactive=True,
            **_profile_actor_kwargs(principal, administrative_visibility=True),
        )
    }
    if profile_id and profile_id not in visible_profile_ids:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Mail-server profile not found",
        )
    rows = list_pop3_imports(
        session,
        tenant_id=principal.tenant_id,
        profile_id=profile_id,
        profile_ids=visible_profile_ids,
        limit=limit,
    )
    return MailPop3ImportListResponse(
        imports=[
            MailPop3ImportRecordResponse.model_validate(pop3_import_payload(row))
            for row in rows
        ]
    )


@router.post("/profiles/{profile_id}/list-imap-folders", response_model=MailImapFolderListResponse)
def list_profile_imap_folders(
    profile_id: str,
    server_id: str | None = Query(default=None),
    credential_id: str | None = Query(default=None),
    campaign_id: str | None = Query(default=None),
    principal: ApiPrincipal = Depends(get_api_principal),
    session: Session = Depends(get_session),
):
    _require_scope(principal, "mail:profile:test")
    _require_scope(principal, "mail:profile:use")
    try:
        profile = _get_profile_for_principal(
            session,
            principal=principal,
            profile_id=profile_id,
            require_active=True,
        )
        if callable(getattr(session, "execute", None)):
            imap = resolve_mail_transport(
                session,
                profile=profile,
                protocol="imap",
                context=_transport_context_for_principal(
                    session,
                    principal=principal,
                    campaign_id=campaign_id,
                ),
                server_id=server_id,
                credential_id=credential_id,
            ).config
        else:
            imap = imap_config_from_profile(profile)
            if imap is None:
                raise MailProfileError("Mail-server profile has no IMAP configuration")
        result = list_imap_folders(imap_config=imap)
        return _mailbox_folder_response(result)
    except (MailProfileError, MailServerHierarchyError) as exc:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
    except Exception as exc:
        return MailImapFolderListResponse(ok=False, message=_safe_error_message(exc), folders=[], detected_sent_folder=None, details={"error_type": exc.__class__.__name__})


@router.get("/profiles/{profile_id}/mailbox/folders", response_model=MailImapFolderListResponse)
def list_profile_mailbox_folders(
    profile_id: str,
    include_status: bool = Query(default=False),
    refresh: bool = False,
    protocol: str = Query(default="imap"),
    principal: ApiPrincipal = Depends(get_api_principal),
    session: Session = Depends(get_session),
):
    _require_mailbox_read_scope(principal)
    mailbox_protocol = _require_mailbox_protocol(protocol)
    try:
        if mailbox_protocol == "jmap":
            jmap = _jmap_config_for_principal(
                session,
                principal=principal,
                profile_id=profile_id,
            )
            return _mailbox_folder_response(
                list_jmap_folders(jmap_config=jmap)
            )
        imap = _imap_config_for_principal(session, principal=principal, profile_id=profile_id)
        if not include_status and not refresh:
            cached = cached_mailbox_folders(session, tenant_id=principal.tenant_id, profile_id=profile_id)
            if cached is not None and not cached.stale:
                result = SimpleNamespace(
                    host=imap.host,
                    port=imap.port,
                    security=imap.security.value,
                    folders=cached.folders,
                    detected_sent_folder=None,
                )
                return _mailbox_folder_response(result, from_cache=True, refreshing=False, indexed_at=cached.indexed_at)
        recovery = begin_mailbox_refresh_recovery(
            tenant_id=principal.tenant_id,
            profile_id=profile_id,
            folder="*",
            purpose="folders",
        )
        try:
            result = list_imap_folders(imap_config=imap, include_status=include_status)
            cache_mailbox_folders(session, tenant_id=principal.tenant_id, profile_id=profile_id, result=result)
            session.commit()
        except Exception as exc:
            session.rollback()
            if not recovery.operation.closed:
                recovery.reject(
                    summary="The read-only IMAP folder refresh did not commit",
                    code=exc.__class__.__name__,
                )
            raise
        recovery.complete_folders(result)
        return _mailbox_folder_response(result)
    except MailboxRefreshBusy as exc:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc
    except MailRecoveryError as exc:
        raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=str(exc)) from exc
    except MailProfileError as exc:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
    except ImapConfigurationError as exc:
        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc
    except ImapAppendError as exc:
        raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc)) from exc
    except (JmapConfigurationError, JmapCapabilityError) as exc:
        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc
    except (JmapAuthenticationError, JmapPermissionError) as exc:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(exc)) from exc
    except JmapProviderError as exc:
        raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc)) from exc


@router.get("/profiles/{profile_id}/mailbox/bootstrap", response_model=MailMailboxBootstrapResponse)
def bootstrap_profile_mailbox(
    profile_id: str,
    folder: str = Query(default="INBOX", min_length=1, max_length=255),
    limit: int = Query(default=DEFAULT_MAILBOX_MESSAGE_LIMIT, ge=1, le=100),
    offset: int = Query(default=0, ge=0, le=100000),
    refresh: bool = False,
    include_status: bool = Query(default=False),
    protocol: str = Query(default="imap"),
    principal: ApiPrincipal = Depends(get_api_principal),
    session: Session = Depends(get_session),
) -> MailMailboxBootstrapResponse:
    _require_mailbox_read_scope(principal)
    mailbox_protocol = _require_mailbox_protocol(protocol)
    try:
        if mailbox_protocol == "jmap":
            jmap = _jmap_config_for_principal(
                session,
                principal=principal,
                profile_id=profile_id,
            )
            result = load_jmap_mailbox_bootstrap(
                jmap_config=jmap,
                folder=folder,
                limit=limit,
                offset=offset,
            )
            next_cursor, cursor_stable = _next_jmap_mailbox_cursor(
                tenant_id=principal.tenant_id,
                profile_id=profile_id,
                folder=result.messages.folder,
                limit=result.messages.limit,
                offset=result.messages.offset,
                query_state=result.messages.uidvalidity,
                messages=result.messages.messages,
                total_count=result.messages.total_count,
                query="",
            )
            return MailMailboxBootstrapResponse(
                profile_id=profile_id,
                folder=result.messages.folder,
                folders=_mailbox_folder_response(result.folders),
                messages=_mailbox_messages_response(
                    profile_id=profile_id,
                    folder=result.messages.folder,
                    host=result.messages.host,
                    port=result.messages.port,
                    security=result.messages.security,
                    total_count=result.messages.total_count,
                    offset=result.messages.offset,
                    limit=result.messages.limit,
                    cursor=None,
                    next_cursor=next_cursor,
                    cursor_stable=cursor_stable,
                    full=False,
                    messages=result.messages.messages,
                ),
            )
        imap = _imap_config_for_principal(session, principal=principal, profile_id=profile_id)
        if not refresh:
            cached_folders = cached_mailbox_folders(session, tenant_id=principal.tenant_id, profile_id=profile_id)
            if cached_folders is not None and not cached_folders.stale:
                selected_folder = _cached_folder_selection(cached_folders.folders, folder)
                cached_messages = cached_mailbox_message_page(
                    session,
                    tenant_id=principal.tenant_id,
                    profile_id=profile_id,
                    folder=selected_folder,
                    limit=limit,
                    offset=offset,
                )
                if cached_messages is not None and not cached_messages.stale:
                    folder_result = SimpleNamespace(
                        host=imap.host,
                        port=imap.port,
                        security=imap.security.value,
                        folders=cached_folders.folders,
                        detected_sent_folder=None,
                    )
                    next_cursor, cursor_stable = _next_mailbox_cursor(
                        tenant_id=principal.tenant_id,
                        profile_id=profile_id,
                        folder=cached_messages.folder,
                        limit=cached_messages.limit,
                        offset=cached_messages.offset,
                        total_count=cached_messages.total_count,
                        uidvalidity=cached_messages.uidvalidity,
                        messages=cached_messages.messages,
                    )
                    return MailMailboxBootstrapResponse(
                        profile_id=profile_id,
                        folder=cached_messages.folder,
                        folders=_mailbox_folder_response(
                            folder_result,
                            from_cache=True,
                            refreshing=False,
                            indexed_at=cached_folders.indexed_at,
                        ),
                        messages=_mailbox_messages_response(
                            profile_id=profile_id,
                            folder=cached_messages.folder,
                            host=imap.host,
                            port=imap.port,
                            security=imap.security.value,
                            total_count=cached_messages.total_count,
                            offset=cached_messages.offset,
                            limit=cached_messages.limit,
                            cursor=None,
                            next_cursor=next_cursor,
                            cursor_stable=cursor_stable,
                            full=not cursor_stable,
                            messages=cached_messages.messages,
                            from_cache=True,
                            refreshing=False,
                            indexed_at=cached_messages.indexed_at,
                        ),
                    )

        recovery = begin_mailbox_refresh_recovery(
            tenant_id=principal.tenant_id,
            profile_id=profile_id,
            folder=folder,
            purpose="bootstrap",
        )
        try:
            result = load_imap_mailbox_bootstrap(
                imap_config=imap,
                folder=folder,
                limit=limit,
                offset=offset,
                include_folder_status=include_status,
            )
            cache_mailbox_folders(session, tenant_id=principal.tenant_id, profile_id=profile_id, result=result.folders)
            cache_mailbox_messages(session, tenant_id=principal.tenant_id, profile_id=profile_id, result=result.messages)
            session.commit()
        except Exception as exc:
            session.rollback()
            if not recovery.operation.closed:
                recovery.reject(
                    summary="The read-only IMAP bootstrap did not commit",
                    code=exc.__class__.__name__,
                )
            raise
        recovery.complete_bootstrap(result.folders, result.messages)
        next_cursor, cursor_stable = _next_mailbox_cursor(
            tenant_id=principal.tenant_id,
            profile_id=profile_id,
            folder=result.messages.folder,
            limit=result.messages.limit,
            offset=result.messages.offset,
            total_count=result.messages.total_count,
            uidvalidity=result.messages.uidvalidity,
            messages=result.messages.messages,
        )
        return MailMailboxBootstrapResponse(
            profile_id=profile_id,
            folder=result.messages.folder,
            folders=_mailbox_folder_response(result.folders),
            messages=_mailbox_messages_response(
                profile_id=profile_id,
                folder=result.messages.folder,
                host=result.messages.host,
                port=result.messages.port,
                security=result.messages.security,
                total_count=result.messages.total_count,
                offset=result.messages.offset,
                limit=result.messages.limit,
                cursor=None,
                next_cursor=next_cursor,
                cursor_stable=cursor_stable,
                full=not cursor_stable,
                messages=result.messages.messages,
            ),
        )
    except MailboxRefreshBusy as exc:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc
    except MailRecoveryError as exc:
        raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=str(exc)) from exc
    except MailProfileError as exc:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
    except ImapConfigurationError as exc:
        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc
    except ImapAppendError as exc:
        raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc)) from exc
    except (JmapConfigurationError, JmapCapabilityError) as exc:
        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc
    except (JmapAuthenticationError, JmapPermissionError) as exc:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(exc)) from exc
    except JmapProviderError as exc:
        raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc)) from exc


@router.get("/profiles/{profile_id}/mailbox/messages", response_model=MailMailboxMessageListResponse)
def list_profile_mailbox_messages(
    profile_id: str,
    folder: str = Query(default="INBOX", min_length=1, max_length=255),
    limit: int | None = Query(default=None, ge=1, le=100),
    offset: int = Query(default=0, ge=0, le=100000),
    cursor: str | None = None,
    refresh: bool = False,
    protocol: str = Query(default="imap"),
    q: str | None = Query(default=None, max_length=500),
    principal: ApiPrincipal = Depends(get_api_principal),
    session: Session = Depends(get_session),
):
    _require_mailbox_read_scope(principal)
    mailbox_protocol = _require_mailbox_protocol(protocol)
    try:
        if mailbox_protocol == "jmap":
            clean_query = q.strip() if isinstance(q, str) else ""
            effective_limit = _jmap_mailbox_cursor_limit(cursor, limit)
            fingerprint = _jmap_mailbox_cursor_fingerprint(
                tenant_id=principal.tenant_id,
                profile_id=profile_id,
                folder=folder,
                limit=effective_limit,
                query=clean_query,
            )
            effective_offset, expected_query_state = _jmap_mailbox_cursor_position(
                cursor,
                fingerprint=fingerprint,
                offset=offset,
            )
            jmap = _jmap_config_for_principal(
                session,
                principal=principal,
                profile_id=profile_id,
            )
            result = list_jmap_messages(
                jmap_config=jmap,
                folder=folder,
                limit=effective_limit,
                offset=effective_offset,
                expected_query_state=expected_query_state,
                query=clean_query,
            )
            next_cursor, cursor_stable = _next_jmap_mailbox_cursor(
                tenant_id=principal.tenant_id,
                profile_id=profile_id,
                folder=result.folder,
                limit=result.limit,
                offset=result.offset,
                total_count=result.total_count,
                query_state=result.uidvalidity,
                messages=result.messages,
                query=clean_query,
            )
            return _mailbox_messages_response(
                profile_id=profile_id,
                folder=result.folder,
                host=result.host,
                port=result.port,
                security=result.security,
                total_count=result.total_count,
                offset=result.offset,
                limit=result.limit,
                cursor=cursor,
                next_cursor=next_cursor,
                cursor_stable=cursor_stable,
                full=result.cursor_reset,
                messages=result.messages,
            )
        imap = _imap_config_for_principal(session, principal=principal, profile_id=profile_id)
        effective_limit = _mailbox_cursor_limit(cursor, limit)
        fingerprint = _mailbox_cursor_fingerprint(tenant_id=principal.tenant_id, profile_id=profile_id, folder=folder, limit=effective_limit)
        effective_offset, after_uid, expected_uidvalidity, cursor_values = _mailbox_cursor_position(cursor, fingerprint=fingerprint, offset=offset)
        if not refresh:
            cached = cached_mailbox_message_page(
                session,
                tenant_id=principal.tenant_id,
                profile_id=profile_id,
                folder=folder,
                limit=effective_limit,
                offset=effective_offset,
            )
            cache_matches_cursor = cached is not None and (
                cursor is None or (expected_uidvalidity is not None and cached.uidvalidity == expected_uidvalidity)
            )
            if cached is not None and cache_matches_cursor and not cached.stale:
                next_cursor, cursor_stable = _next_mailbox_cursor(
                    tenant_id=principal.tenant_id,
                    profile_id=profile_id,
                    folder=cached.folder,
                    limit=cached.limit,
                    offset=cached.offset,
                    total_count=cached.total_count,
                    uidvalidity=cached.uidvalidity,
                    messages=cached.messages,
                )
                return _mailbox_messages_response(
                    profile_id=profile_id,
                    folder=cached.folder,
                    host=imap.host,
                    port=imap.port,
                    security=imap.security.value,
                    total_count=cached.total_count,
                    offset=cached.offset,
                    limit=cached.limit,
                    cursor=cursor,
                    next_cursor=next_cursor,
                    cursor_stable=cursor_stable,
                    full=not cursor_stable,
                    messages=cached.messages,
                    from_cache=True,
                    refreshing=False,
                    indexed_at=cached.indexed_at,
                )
        recovery = begin_mailbox_refresh_recovery(
            tenant_id=principal.tenant_id,
            profile_id=profile_id,
            folder=folder,
            purpose="messages",
        )
        try:
            result = list_imap_messages(
                imap_config=imap,
                folder=folder,
                limit=effective_limit,
                offset=effective_offset,
                after_uid=after_uid,
                expected_uidvalidity=expected_uidvalidity,
            )
            cache_mailbox_messages(session, tenant_id=principal.tenant_id, profile_id=profile_id, result=result)
            session.commit()
        except Exception as exc:
            session.rollback()
            if not recovery.operation.closed:
                recovery.reject(
                    summary="The read-only IMAP message refresh did not commit",
                    code=exc.__class__.__name__,
                )
            raise
        recovery.complete_messages(result)
        full = bool(cursor_values is not None and result.cursor_reset)
        next_cursor, cursor_stable = _next_mailbox_cursor(
            tenant_id=principal.tenant_id,
            profile_id=profile_id,
            folder=result.folder,
            limit=result.limit,
            offset=result.offset,
            total_count=result.total_count,
            uidvalidity=result.uidvalidity,
            messages=result.messages,
        )
        return _mailbox_messages_response(
            profile_id=profile_id,
            folder=result.folder,
            host=result.host,
            port=result.port,
            security=result.security,
            total_count=result.total_count,
            offset=result.offset,
            limit=result.limit,
            cursor=cursor,
            next_cursor=next_cursor,
            cursor_stable=cursor_stable,
            full=full or not cursor_stable,
            messages=result.messages,
        )
    except MailboxRefreshBusy as exc:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc
    except MailRecoveryError as exc:
        raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=str(exc)) from exc
    except MailProfileError as exc:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
    except ImapConfigurationError as exc:
        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc
    except ImapAppendError as exc:
        raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc)) from exc
    except (JmapConfigurationError, JmapCapabilityError) as exc:
        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc
    except (JmapAuthenticationError, JmapPermissionError) as exc:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(exc)) from exc
    except JmapProviderError as exc:
        raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc)) from exc


@router.get(
    "/profiles/{profile_id}/mailbox/changes",
    response_model=MailMailboxChangesResponse,
)
def get_profile_mailbox_changes(
    profile_id: str,
    since_state: str = Query(min_length=1, max_length=1_000),
    max_changes: int = Query(default=500, ge=1, le=1_000),
    principal: ApiPrincipal = Depends(get_api_principal),
    session: Session = Depends(get_session),
):
    _require_mailbox_read_scope(principal)
    try:
        jmap = _jmap_config_for_principal(
            session,
            principal=principal,
            profile_id=profile_id,
        )
        result = get_jmap_email_changes(
            jmap_config=jmap,
            since_state=since_state,
            max_changes=max_changes,
        )
        return MailMailboxChangesResponse(
            profile_id=profile_id,
            account_id=result.account_id,
            old_state=result.old_state,
            new_state=result.new_state,
            has_more_changes=result.has_more_changes,
            created=list(result.created),
            updated=list(result.updated),
            destroyed=list(result.destroyed),
        )
    except MailProfileError as exc:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
    except (JmapConfigurationError, JmapCapabilityError) as exc:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc
    except (JmapAuthenticationError, JmapPermissionError) as exc:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(exc)) from exc
    except JmapProviderError as exc:
        raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc)) from exc


@router.get("/profiles/{profile_id}/mailbox/messages/{message_uid}", response_model=MailMailboxMessageResponse)
def get_profile_mailbox_message(
    profile_id: str,
    message_uid: str,
    folder: str = Query(default="INBOX", min_length=1, max_length=255),
    protocol: str = Query(default="imap"),
    principal: ApiPrincipal = Depends(get_api_principal),
    session: Session = Depends(get_session),
):
    _require_mailbox_read_scope(principal)
    mailbox_protocol = _require_mailbox_protocol(protocol)
    try:
        if mailbox_protocol == "jmap":
            jmap = _jmap_config_for_principal(
                session,
                principal=principal,
                profile_id=profile_id,
            )
            result = get_jmap_message(
                jmap_config=jmap,
                folder=folder,
                email_id=message_uid,
            )
            return MailMailboxMessageResponse(
                profile_id=profile_id,
                folder=result.folder,
                host=result.host,
                port=result.port,
                security=result.security,
                message=_mailbox_detail_response(result.message),
            )
        imap = _imap_config_for_principal(session, principal=principal, profile_id=profile_id)
        result = get_imap_message(imap_config=imap, folder=folder, uid=message_uid)
        return MailMailboxMessageResponse(
            profile_id=profile_id,
            folder=result.folder,
            host=result.host,
            port=result.port,
            security=result.security,
            message=_mailbox_detail_response(result.message),
        )
    except MailProfileError as exc:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
    except ImapConfigurationError as exc:
        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc
    except ImapAppendError as exc:
        status_code = status.HTTP_404_NOT_FOUND if "not found" in str(exc).casefold() else status.HTTP_502_BAD_GATEWAY
        raise HTTPException(status_code=status_code, detail=str(exc)) from exc
    except (JmapConfigurationError, JmapCapabilityError) as exc:
        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc
    except (JmapAuthenticationError, JmapPermissionError) as exc:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(exc)) from exc
    except JmapProviderError as exc:
        status_code = status.HTTP_404_NOT_FOUND if "not found" in str(exc).casefold() else status.HTTP_502_BAD_GATEWAY
        raise HTTPException(status_code=status_code, detail=str(exc)) from exc


def _safe_error_message(exc: Exception) -> str:
    text = str(exc).strip()
    return text or exc.__class__.__name__


@router.post("/test-smtp", response_model=MailConnectionTestResponse)
def test_smtp_settings(
    payload: MailSmtpTestRequest,
    principal: ApiPrincipal = Depends(get_api_principal),
):
    """Test SMTP connectivity/login without sending any message."""

    if not has_scope(principal, "mail:profile:test"):
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Missing scope: mail_servers:test")
    if not has_scope(principal, "mail:secret:manage"):
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Testing raw credentials requires mail_servers:manage_credentials")
    try:
        result = test_smtp_login(smtp_config=payload)
        return MailConnectionTestResponse(
            ok=True,
            protocol="smtp",
            host=result.host,
            port=result.port,
            security=result.security,
            message="SMTP connection successful.",
            details={"authenticated": result.authenticated},
        )
    except Exception as exc:
        return MailConnectionTestResponse(
            ok=False,
            protocol="smtp",
            host=payload.host,
            port=payload.port,
            security=payload.security.value,
            message=_safe_error_message(exc),
            details={"error_type": exc.__class__.__name__},
        )


@router.post("/test-imap", response_model=MailConnectionTestResponse)
def test_imap_settings(
    payload: MailImapTestRequest,
    principal: ApiPrincipal = Depends(get_api_principal),
):
    """Test IMAP connectivity/login without selecting or appending messages."""

    if not has_scope(principal, "mail:profile:test"):
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Missing scope: mail_servers:test")
    if not has_scope(principal, "mail:secret:manage"):
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Testing raw credentials requires mail_servers:manage_credentials")
    try:
        result = test_imap_login(imap_config=payload)
        return MailConnectionTestResponse(
            ok=True,
            protocol="imap",
            host=result.host,
            port=result.port,
            security=result.security,
            message="IMAP connection successful.",
            details={"authenticated": result.authenticated},
        )
    except Exception as exc:
        return MailConnectionTestResponse(
            ok=False,
            protocol="imap",
            host=payload.host,
            port=payload.port,
            security=payload.security.value,
            message=_safe_error_message(exc),
            details={"error_type": exc.__class__.__name__},
        )


@router.post("/list-imap-folders", response_model=MailImapFolderListResponse)
def list_imap_folder_settings(
    payload: MailImapTestRequest,
    principal: ApiPrincipal = Depends(get_api_principal),
):
    """List visible IMAP folders and return the best Sent-folder guess."""

    if not has_scope(principal, "mail:profile:test"):
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Missing scope: mail_servers:test")
    if not has_scope(principal, "mail:secret:manage"):
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Testing raw credentials requires mail_servers:manage_credentials")
    try:
        result = list_imap_folders(imap_config=payload)
        folders = [MailImapFolderResponse(name=item.name, flags=item.flags, message_count=item.message_count, unseen_count=item.unseen_count) for item in result.folders]
        return MailImapFolderListResponse(
            ok=True,
            host=result.host,
            port=result.port,
            security=result.security,
            message=f"Found {len(folders)} IMAP folder(s).",
            folders=folders,
            detected_sent_folder=result.detected_sent_folder,
            detected_folder_mappings=getattr(result, "detected_folder_mappings", None) or {},
        )
    except Exception as exc:
        return MailImapFolderListResponse(
            ok=False,
            host=payload.host,
            port=payload.port,
            security=payload.security.value,
            message=_safe_error_message(exc),
            folders=[],
            detected_sent_folder=None,
            details={"error_type": exc.__class__.__name__},
        )