from __future__ import annotations

from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from datetime import datetime, timezone

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

from govoplan_addresses.backend.db.models import (
    AddressBook,
    AddressImportProfile,
    AddressImportRun,
    AddressList,
    AddressListEntry,
    AddressSyncConflict,
    AddressSyncSource,
    AddressSyncTombstone,
    Contact,
    ContactChannelRule,
    ContactEmail,
    ContactFieldProvenance,
    ContactMergeRecord,
    ContactPhone,
    ContactPointQualityDecision,
    ContactPointSnapshot,
    ContactPostalAddress,
    ContactRedirect,
)
from govoplan_core.core.dsar import (
    DsarErasureActionRef,
    DsarExecutionResultRef,
    DsarRecordRef,
    DsarSubjectRef,
    dsar_capability_name,
)


ADDRESSES_DSAR_CAPABILITY = dsar_capability_name("addresses")
_MAX_RECORDS = 5_000


@dataclass(frozen=True, slots=True)
class _SubjectSelectors:
    account_id: str | None
    email: str | None
    references: Mapping[str, str]


class AddressesDsarProvider:
    provider_id = "addresses"
    module_id = "addresses"

    def search_subject(
        self,
        session: object,
        *,
        tenant_id: str,
        subject: DsarSubjectRef,
    ) -> Sequence[DsarRecordRef]:
        db = _session(session)
        selectors = _subject_selectors(subject)
        if selectors is None:
            return ()

        contact_ids = _subject_contact_ids(
            db,
            tenant_id=tenant_id,
            selectors=selectors,
        )
        if contact_ids is None:
            return ()

        records: list[DsarRecordRef] = []
        seen: set[tuple[str, str]] = set()

        def append(record: DsarRecordRef) -> None:
            key = (record.resource_type, record.resource_id)
            if key in seen:
                return
            if len(records) >= _MAX_RECORDS:
                raise ValueError(
                    "Addresses DSAR result limit exceeded; narrow the subject selectors."
                )
            seen.add(key)
            records.append(record)

        contacts = _rows_by_ids(
            db,
            Contact,
            tenant_id=tenant_id,
            ids=contact_ids,
        )
        for contact in contacts:
            match_fields = []
            if selectors.references.get("contact") == contact.id:
                match_fields.append("reference")
            if selectors.email and any(
                _normalized_email(item.email) == selectors.email
                or _normalized_email(item.normalized_email) == selectors.email
                for item in contact.emails
            ):
                match_fields.append("email")
            append(
                _record(
                    "addresses_contact",
                    contact.id,
                    "contact_identity",
                    f"Address contact {contact.display_name}",
                    {
                        "match_fields": match_fields,
                        "display_name": _bounded_text(contact.display_name, 255),
                        "given_name": _bounded_text(contact.given_name, 255),
                        "family_name": _bounded_text(contact.family_name, 255),
                        "organization": _bounded_text(contact.organization, 255),
                        "role_title": _bounded_text(contact.role_title, 255),
                        "note": _bounded_text(contact.note, 2_000),
                        "tags": [str(value)[:120] for value in contact.tags[:100]],
                        "source_kind": contact.source_kind,
                        "source_revision": _bounded_text(
                            contact.source_revision,
                            255,
                        ),
                        "deleted_at": _iso(contact.deleted_at),
                    },
                    observed_at=contact.updated_at,
                )
            )

        for email in _contact_children(
            db,
            ContactEmail,
            tenant_id=tenant_id,
            contact_ids=contact_ids,
        ):
            append(
                _record(
                    "addresses_contact_email",
                    email.id,
                    "electronic_contact_point",
                    "Address email contact point",
                    {
                        "contact_id": email.contact_id,
                        "match_fields": _contact_point_match_fields(
                            email,
                            selectors=selectors,
                            reference_kind="contact_email",
                        ),
                        "label": _bounded_text(email.label, 80),
                        "email": _bounded_text(email.email, 320),
                        "original_email": _bounded_text(email.original_email, 320),
                        "normalized_email": _bounded_text(
                            email.normalized_email,
                            320,
                        ),
                        "is_primary": email.is_primary,
                        "order_index": email.order_index,
                    },
                    observed_at=email.updated_at,
                )
            )

        for phone in _contact_children(
            db,
            ContactPhone,
            tenant_id=tenant_id,
            contact_ids=contact_ids,
        ):
            append(
                _record(
                    "addresses_contact_phone",
                    phone.id,
                    "telephone_contact_point",
                    "Address telephone contact point",
                    {
                        "contact_id": phone.contact_id,
                        "match_fields": _contact_point_match_fields(
                            phone,
                            selectors=selectors,
                            reference_kind="contact_phone",
                        ),
                        "label": _bounded_text(phone.label, 80),
                        "phone": _bounded_text(phone.phone, 100),
                        "original_phone": _bounded_text(phone.original_phone, 100),
                        "normalized_phone": _bounded_text(
                            phone.normalized_phone,
                            100,
                        ),
                        "is_primary": phone.is_primary,
                        "order_index": phone.order_index,
                    },
                    observed_at=phone.updated_at,
                )
            )

        for postal in _contact_children(
            db,
            ContactPostalAddress,
            tenant_id=tenant_id,
            contact_ids=contact_ids,
        ):
            append(
                _record(
                    "addresses_contact_postal_address",
                    postal.id,
                    "postal_contact_point",
                    "Address postal contact point",
                    {
                        "contact_id": postal.contact_id,
                        "match_fields": _contact_point_match_fields(
                            postal,
                            selectors=selectors,
                            reference_kind="contact_postal_address",
                        ),
                        "label": _bounded_text(postal.label, 80),
                        "street": _bounded_text(postal.street, 500),
                        "postal_code": _bounded_text(postal.postal_code, 40),
                        "locality": _bounded_text(postal.locality, 255),
                        "region": _bounded_text(postal.region, 255),
                        "country": _bounded_text(postal.country, 255),
                        "is_primary": postal.is_primary,
                        "order_index": postal.order_index,
                    },
                    observed_at=postal.updated_at,
                )
            )

        for entry in _list_entries(
            db,
            tenant_id=tenant_id,
            contact_ids=contact_ids,
        ):
            append(
                _record(
                    "addresses_list_membership",
                    entry.id,
                    "address_list_use",
                    "Address-list membership",
                    {
                        "contact_id": entry.contact_id,
                        "address_list_id": entry.address_list_id,
                        "address_list_name": _bounded_text(
                            entry.address_list.name,
                            255,
                        ),
                        "target_kind": entry.target_kind,
                        "contact_email_id": entry.contact_email_id,
                        "contact_postal_address_id": (entry.contact_postal_address_id),
                        "label": _bounded_text(entry.label, 255),
                        "order_index": entry.order_index,
                    },
                    observed_at=entry.updated_at,
                )
            )

        for rule in _related_or_attributed(
            db,
            ContactChannelRule,
            tenant_id=tenant_id,
            related_field=ContactChannelRule.contact_id,
            related_ids=contact_ids,
            account_id=selectors.account_id,
            actor_fields=(ContactChannelRule.created_by_account_id,),
        ):
            append(
                _record(
                    "addresses_channel_rule",
                    rule.id,
                    "communication_governance_evidence",
                    "Address communication-governance decision",
                    {
                        "match_fields": _related_actor_match_fields(
                            rule,
                            contact_ids=contact_ids,
                            account_id=selectors.account_id,
                            actor_fields=("created_by_account_id",),
                        ),
                        "contact_id": (
                            rule.contact_id if rule.contact_id in contact_ids else None
                        ),
                        "channel": rule.channel,
                        "purpose": _bounded_text(rule.purpose, 120),
                        "contact_point_id": (
                            rule.contact_point_id
                            if rule.contact_id in contact_ids
                            else None
                        ),
                        "decision": rule.decision,
                        "legal_basis": _bounded_text(rule.legal_basis, 255),
                        "reason": _bounded_text(rule.reason, 2_000),
                        "preference_rank": rule.preference_rank,
                        "locale": _bounded_text(rule.locale, 20),
                        "effective_from": _iso(rule.effective_from),
                        "effective_until": _iso(rule.effective_until),
                    },
                    observed_at=rule.updated_at,
                    immutable=True,
                    retention_reason=(
                        "Communication-governance decisions are effective-dated "
                        "institutional evidence and must be corrected by a new decision."
                    ),
                )
            )

        for decision in _related_or_attributed(
            db,
            ContactPointQualityDecision,
            tenant_id=tenant_id,
            related_field=ContactPointQualityDecision.contact_id,
            related_ids=contact_ids,
            account_id=selectors.account_id,
            actor_fields=(ContactPointQualityDecision.created_by_account_id,),
        ):
            append(
                _record(
                    "addresses_quality_decision",
                    decision.id,
                    "contact_quality_evidence",
                    "Address contact-point quality decision",
                    {
                        "match_fields": _related_actor_match_fields(
                            decision,
                            contact_ids=contact_ids,
                            account_id=selectors.account_id,
                            actor_fields=("created_by_account_id",),
                        ),
                        "contact_id": (
                            decision.contact_id
                            if decision.contact_id in contact_ids
                            else None
                        ),
                        "channel": decision.channel,
                        "contact_point_id": (
                            decision.contact_point_id
                            if decision.contact_id in contact_ids
                            else None
                        ),
                        "state": decision.state,
                        "reason_code": decision.reason_code,
                        "reason": _bounded_text(decision.reason, 2_000),
                        "effective_from": _iso(decision.effective_from),
                        "effective_until": _iso(decision.effective_until),
                    },
                    observed_at=decision.updated_at,
                    immutable=True,
                    retention_reason=(
                        "Quality decisions explain recipient eligibility and are "
                        "retained as effective-dated evidence."
                    ),
                )
            )

        for provenance in _related_or_attributed(
            db,
            ContactFieldProvenance,
            tenant_id=tenant_id,
            related_field=ContactFieldProvenance.contact_id,
            related_ids=contact_ids,
            account_id=selectors.account_id,
            actor_fields=(ContactFieldProvenance.created_by_account_id,),
        ):
            append(
                _record(
                    "addresses_field_provenance",
                    provenance.id,
                    "contact_provenance_evidence",
                    "Address field provenance",
                    {
                        "match_fields": _related_actor_match_fields(
                            provenance,
                            contact_ids=contact_ids,
                            account_id=selectors.account_id,
                            actor_fields=("created_by_account_id",),
                        ),
                        "contact_id": (
                            provenance.contact_id
                            if provenance.contact_id in contact_ids
                            else None
                        ),
                        "field_path": _bounded_text(provenance.field_path, 255),
                        "source_kind": provenance.source_kind,
                        "source_revision": _bounded_text(
                            provenance.source_revision,
                            255,
                        ),
                        "precedence": provenance.precedence,
                        "selected": provenance.selected,
                        "reason_code": provenance.reason_code,
                        "explanation": _bounded_text(
                            provenance.explanation,
                            2_000,
                        ),
                        "visibility": provenance.visibility,
                    },
                    observed_at=provenance.updated_at,
                    immutable=True,
                    retention_reason=(
                        "Field provenance is retained to explain source authority, "
                        "normalization, and merge survivorship."
                    ),
                )
            )

        for merge in _matching_merges(
            db,
            tenant_id=tenant_id,
            contact_ids=contact_ids,
            account_id=selectors.account_id,
        ):
            related_ids = {
                merge.winner_contact_id,
                *(str(item) for item in merge.loser_contact_ids),
            } & contact_ids
            actor_fields = _actor_match_fields(
                merge,
                selectors.account_id,
                (
                    "created_by_account_id",
                    "recovered_by_account_id",
                ),
            )
            append(
                _record(
                    "addresses_merge_record",
                    merge.id,
                    "contact_merge_evidence",
                    "Address contact merge record",
                    {
                        "match_fields": (
                            (["contact_id"] if related_ids else []) + actor_fields
                        ),
                        "related_contact_ids": sorted(related_ids),
                        "status": merge.status,
                        "reason": _bounded_text(merge.reason, 2_000),
                        "before_hash": merge.before_hash,
                        "after_hash": merge.after_hash,
                        "recovered_at": _iso(merge.recovered_at),
                        "recovery_action": merge.recovery_action,
                        "recovery_reason": _bounded_text(
                            merge.recovery_reason,
                            2_000,
                        ),
                    },
                    observed_at=merge.updated_at,
                    immutable=True,
                    retention_reason=(
                        "Merge records and hashes are retained so redirects, list "
                        "repairs, undo, and split operations remain explainable."
                    ),
                )
            )

        for redirect in _matching_redirects(
            db,
            tenant_id=tenant_id,
            contact_ids=contact_ids,
        ):
            append(
                _record(
                    "addresses_contact_redirect",
                    redirect.id,
                    "contact_merge_evidence",
                    "Address contact redirect",
                    {
                        "source_contact_id": redirect.source_contact_id,
                        "target_contact_id": redirect.target_contact_id,
                        "merge_record_id": redirect.merge_record_id,
                        "ended_at": _iso(redirect.ended_at),
                    },
                    observed_at=redirect.updated_at,
                    immutable=True,
                    retention_reason=(
                        "Contact redirects preserve resolution of historical references."
                    ),
                )
            )

        for tombstone in _related_rows(
            db,
            AddressSyncTombstone,
            tenant_id=tenant_id,
            related_field=AddressSyncTombstone.contact_id,
            related_ids=contact_ids,
        ):
            append(
                _record(
                    "addresses_sync_tombstone",
                    tombstone.id,
                    "address_sync_evidence",
                    "Address synchronization tombstone",
                    {
                        "contact_id": tombstone.contact_id,
                        "sync_source_id": tombstone.sync_source_id,
                        "address_book_id": tombstone.address_book_id,
                        "local_deleted_at": _iso(tombstone.local_deleted_at),
                        "remote_deleted_at": _iso(tombstone.remote_deleted_at),
                        "synced_at": _iso(tombstone.synced_at),
                    },
                    observed_at=tombstone.updated_at,
                    immutable=True,
                    retention_reason=(
                        "Synchronization tombstones prevent accidental recreation "
                        "and document source lifecycle outcomes."
                    ),
                )
            )

        for conflict in _related_or_attributed(
            db,
            AddressSyncConflict,
            tenant_id=tenant_id,
            related_field=AddressSyncConflict.contact_id,
            related_ids=contact_ids,
            account_id=selectors.account_id,
            actor_fields=(AddressSyncConflict.resolved_by_account_id,),
        ):
            append(
                _record(
                    "addresses_sync_conflict",
                    conflict.id,
                    "address_sync_evidence",
                    "Address synchronization conflict",
                    {
                        "match_fields": _related_actor_match_fields(
                            conflict,
                            contact_ids=contact_ids,
                            account_id=selectors.account_id,
                            actor_fields=("resolved_by_account_id",),
                        ),
                        "contact_id": (
                            conflict.contact_id
                            if conflict.contact_id in contact_ids
                            else None
                        ),
                        "sync_source_id": conflict.sync_source_id,
                        "address_book_id": conflict.address_book_id,
                        "field_path": _bounded_text(conflict.field_path, 500),
                        "status": conflict.status,
                        "resolution": conflict.resolution,
                        "resolved_at": _iso(conflict.resolved_at),
                    },
                    observed_at=conflict.updated_at,
                    immutable=True,
                    retention_reason=(
                        "Synchronization conflict outcomes are retained to explain "
                        "which authoritative value was selected."
                    ),
                )
            )

        _append_actor_attribution_records(
            db,
            tenant_id=tenant_id,
            account_id=selectors.account_id,
            append=append,
        )

        return tuple(records)

    def plan_erasure(
        self,
        session: object,
        *,
        tenant_id: str,
        subject: DsarSubjectRef,
        records: Sequence[DsarRecordRef],
    ) -> Sequence[DsarErasureActionRef]:
        del tenant_id
        _session(session)
        if _subject_selectors(subject) is None:
            raise ValueError("Addresses DSAR subject selectors conflict.")
        actions: list[DsarErasureActionRef] = []
        for record in records:
            _validate_record(record)
            if record.immutable_evidence:
                kind = "retain"
                title = f"Retain {record.title}"
                rationale = record.retention_reason or (
                    "Addresses governance evidence must be retained."
                )
            else:
                kind = "manual_review"
                title = f"Review {record.title}"
                rationale = (
                    "Reusable address data can be shared, synchronized, merged, "
                    "or referenced by immutable recipient evidence. An authorized "
                    "operator must correct, archive, or erase it through the owning "
                    "Addresses workflow after reviewing those dependencies."
                )
            actions.append(
                DsarErasureActionRef(
                    action_id=(
                        f"addresses:{kind}:{record.resource_type}:{record.resource_id}"
                    ),
                    provider_id=self.provider_id,
                    module_id=self.module_id,
                    kind=kind,
                    resource_type=record.resource_type,
                    resource_id=record.resource_id,
                    title=title,
                    rationale=rationale,
                    executable=False,
                )
            )
        return tuple(actions)

    def execute_erasure(
        self,
        session: object,
        *,
        tenant_id: str,
        subject: DsarSubjectRef,
        actions: Sequence[DsarErasureActionRef],
        request_id: str,
    ) -> Sequence[DsarExecutionResultRef]:
        del tenant_id
        _session(session)
        if _subject_selectors(subject) is None:
            raise ValueError("Addresses DSAR subject selectors conflict.")
        results: list[DsarExecutionResultRef] = []
        for action in actions:
            _validate_action(action)
            if action.executable:
                raise ValueError(
                    "Addresses DSAR does not publish executable erasure actions."
                )
            results.append(
                DsarExecutionResultRef(
                    action_id=action.action_id,
                    status="blocked",
                    summary=(
                        "Use the authorized Addresses correction, archive, source, "
                        "merge, or governance workflow after dependency review."
                    ),
                    evidence={"request_id": request_id},
                )
            )
        return tuple(results)


def _subject_contact_ids(
    session: Session,
    *,
    tenant_id: str,
    selectors: _SubjectSelectors,
) -> set[str] | None:
    email_ids: set[str] = set()
    if selectors.email:
        email_ids = {
            row[0]
            for row in _bounded_rows(
                session.query(Contact.id)
                .join(ContactEmail, ContactEmail.contact_id == Contact.id)
                .filter(
                    Contact.tenant_id == tenant_id,
                    or_(
                        func.lower(ContactEmail.email) == selectors.email,
                        func.lower(ContactEmail.normalized_email) == selectors.email,
                    ),
                )
                .order_by(Contact.id)
            )
        }

    direct_ids: set[str] = set()
    direct_queries = (
        ("contact", Contact, Contact.id),
        ("contact_email", ContactEmail, ContactEmail.id),
        ("contact_phone", ContactPhone, ContactPhone.id),
        (
            "contact_postal_address",
            ContactPostalAddress,
            ContactPostalAddress.id,
        ),
    )
    for kind, model, id_field in direct_queries:
        reference = selectors.references.get(kind)
        if not reference:
            continue
        query = session.query(Contact.id)
        if model is not Contact:
            query = query.join(model, model.contact_id == Contact.id)
        row = (
            query.filter(
                Contact.tenant_id == tenant_id,
                id_field == reference,
            )
            .order_by(Contact.id)
            .one_or_none()
        )
        if row is None:
            return None
        direct_ids.add(row[0])

    if selectors.email and direct_ids and not direct_ids.issubset(email_ids):
        return None
    return email_ids | direct_ids


def _append_actor_attribution_records(
    session: Session,
    *,
    tenant_id: str,
    account_id: str | None,
    append: object,
) -> None:
    if not account_id:
        return

    definitions = (
        (
            AddressBook,
            "addresses_address_book_attribution",
            "address_book_configuration",
            ("created_by_account_id", "updated_by_account_id"),
            lambda row: {
                "scope_type": row.scope_type,
                "scope_id": row.scope_id,
                "name": _bounded_text(row.name, 255),
                "source_kind": row.source_kind,
                "read_only": row.read_only,
                "deleted_at": _iso(row.deleted_at),
            },
        ),
        (
            Contact,
            "addresses_contact_attribution",
            "contact_configuration",
            ("created_by_account_id", "updated_by_account_id"),
            lambda row: {
                "address_book_id": row.address_book_id,
                "source_kind": row.source_kind,
                "deleted_at": _iso(row.deleted_at),
            },
        ),
        (
            AddressList,
            "addresses_list_attribution",
            "address_list_configuration",
            ("created_by_account_id", "updated_by_account_id"),
            lambda row: {
                "address_book_id": row.address_book_id,
                "name": _bounded_text(row.name, 255),
                "source_kind": row.source_kind,
                "read_only": row.read_only,
                "deleted_at": _iso(row.deleted_at),
            },
        ),
        (
            AddressSyncSource,
            "addresses_sync_source_attribution",
            "address_source_configuration",
            ("created_by_account_id", "updated_by_account_id"),
            lambda row: {
                "address_book_id": row.address_book_id,
                "connector_type": row.connector_type,
                "display_name": _bounded_text(row.display_name, 255),
                "sync_direction": row.sync_direction,
                "read_only": row.read_only,
                "enabled": row.enabled,
                "status": row.status,
            },
        ),
        (
            AddressImportProfile,
            "addresses_import_profile_attribution",
            "address_import_configuration",
            ("created_by_account_id",),
            lambda row: {
                "profile_key": row.profile_key,
                "version": row.version,
                "scope_type": row.scope_type,
                "scope_id": row.scope_id,
                "name": _bounded_text(row.name, 255),
                "source_format": row.source_format,
                "is_current": row.is_current,
                "superseded_at": _iso(row.superseded_at),
            },
        ),
        (
            AddressImportRun,
            "addresses_import_run_attribution",
            "address_import_evidence",
            ("created_by_account_id",),
            lambda row: {
                "address_book_id": row.address_book_id,
                "profile_id": row.profile_id,
                "source_filename": _bounded_text(row.source_filename, 500),
                "source_format": row.source_format,
                "input_hash": row.input_hash,
                "plan_hash": row.plan_hash,
                "status": row.status,
                "row_count": row.row_count,
                "applied_at": _iso(row.applied_at),
                "rolled_back_at": _iso(row.rolled_back_at),
            },
        ),
        (
            ContactPointSnapshot,
            "addresses_snapshot_attribution",
            "recipient_snapshot_evidence",
            ("created_by_account_id",),
            lambda row: {
                "source_id": row.source_id,
                "contract_version": row.contract_version,
                "source_revision": row.source_revision,
                "purpose": _bounded_text(row.purpose, 120),
                "effective_at": _iso(row.effective_at),
                "generated_at": _iso(row.generated_at),
                "recipient_count": row.recipient_count,
                "excluded_count": row.excluded_count,
                "snapshot_hash": row.snapshot_hash,
            },
        ),
    )
    append_record = append
    for model, resource_type, category, actor_fields, data_factory in definitions:
        conditions = [getattr(model, field) == account_id for field in actor_fields]
        rows = _bounded_rows(
            session.query(model)
            .filter(model.tenant_id == tenant_id, or_(*conditions))
            .order_by(model.id)
        )
        for row in rows:
            data = data_factory(row)
            data["match_fields"] = _actor_match_fields(
                row,
                account_id,
                actor_fields,
            )
            append_record(  # type: ignore[operator]
                _record(
                    resource_type,
                    row.id,
                    category,
                    "Addresses operator attribution",
                    data,
                    observed_at=row.updated_at,
                    immutable=True,
                    retention_reason=(
                        "Operator attribution is retained with the governed "
                        "configuration or evidence record for accountability."
                    ),
                )
            )


def _subject_selectors(subject: DsarSubjectRef) -> _SubjectSelectors | None:
    groups = {
        "account_id": (
            subject.account_id,
            subject.external_references.get("addresses.account"),
            subject.external_references.get("access.account"),
        ),
        "email": (
            subject.email,
            subject.external_references.get("addresses.email"),
        ),
    }
    normalized: dict[str, str | None] = {}
    for key, values in groups.items():
        distinct = {
            value
            for item in values
            if (
                value := (
                    _normalized_email(item) if key == "email" else _normalized_id(item)
                )
            )
        }
        if len(distinct) > 1:
            return None
        normalized[key] = next(iter(distinct), None)

    aliases = {
        "addresses.contact": "contact",
        "addresses.contact_email": "contact_email",
        "addresses.contact_phone": "contact_phone",
        "addresses.contact_postal_address": "contact_postal_address",
    }
    references = {
        target: value
        for source, target in aliases.items()
        if (value := _normalized_id(subject.external_references.get(source)))
    }
    return _SubjectSelectors(references=references, **normalized)


def _rows_by_ids(
    session: Session,
    model: type,
    *,
    tenant_id: str,
    ids: set[str],
) -> list[object]:
    if not ids:
        return []
    return _bounded_rows(
        session.query(model)
        .filter(model.tenant_id == tenant_id, model.id.in_(ids))
        .order_by(model.id)
    )


def _contact_children(
    session: Session,
    model: type,
    *,
    tenant_id: str,
    contact_ids: set[str],
) -> list[object]:
    if not contact_ids:
        return []
    return _bounded_rows(
        session.query(model)
        .join(Contact, model.contact_id == Contact.id)
        .filter(
            Contact.tenant_id == tenant_id,
            model.contact_id.in_(contact_ids),
        )
        .order_by(model.id)
    )


def _list_entries(
    session: Session,
    *,
    tenant_id: str,
    contact_ids: set[str],
) -> list[AddressListEntry]:
    if not contact_ids:
        return []
    return _bounded_rows(
        session.query(AddressListEntry)
        .join(AddressList, AddressListEntry.address_list_id == AddressList.id)
        .filter(
            AddressList.tenant_id == tenant_id,
            AddressListEntry.contact_id.in_(contact_ids),
        )
        .order_by(AddressListEntry.id)
    )


def _related_rows(
    session: Session,
    model: type,
    *,
    tenant_id: str,
    related_field: object,
    related_ids: set[str],
) -> list[object]:
    if not related_ids:
        return []
    return _bounded_rows(
        session.query(model)
        .filter(
            model.tenant_id == tenant_id,
            related_field.in_(related_ids),  # type: ignore[attr-defined]
        )
        .order_by(model.id)
    )


def _related_or_attributed(
    session: Session,
    model: type,
    *,
    tenant_id: str,
    related_field: object,
    related_ids: set[str],
    account_id: str | None,
    actor_fields: Sequence[object],
) -> list[object]:
    conditions = []
    if related_ids:
        conditions.append(related_field.in_(related_ids))  # type: ignore[attr-defined]
    if account_id:
        conditions.extend(field == account_id for field in actor_fields)
    if not conditions:
        return []
    return _bounded_rows(
        session.query(model)
        .filter(model.tenant_id == tenant_id, or_(*conditions))
        .order_by(model.id)
    )


def _matching_merges(
    session: Session,
    *,
    tenant_id: str,
    contact_ids: set[str],
    account_id: str | None,
) -> list[ContactMergeRecord]:
    rows = _bounded_rows(
        session.query(ContactMergeRecord)
        .filter(ContactMergeRecord.tenant_id == tenant_id)
        .order_by(ContactMergeRecord.id)
    )
    return [
        row
        for row in rows
        if row.winner_contact_id in contact_ids
        or bool({str(item) for item in row.loser_contact_ids} & contact_ids)
        or bool(
            _actor_match_fields(
                row,
                account_id,
                ("created_by_account_id", "recovered_by_account_id"),
            )
        )
    ]


def _matching_redirects(
    session: Session,
    *,
    tenant_id: str,
    contact_ids: set[str],
) -> list[ContactRedirect]:
    if not contact_ids:
        return []
    return _bounded_rows(
        session.query(ContactRedirect)
        .filter(
            ContactRedirect.tenant_id == tenant_id,
            or_(
                ContactRedirect.source_contact_id.in_(contact_ids),
                ContactRedirect.target_contact_id.in_(contact_ids),
            ),
        )
        .order_by(ContactRedirect.id)
    )


def _contact_point_match_fields(
    row: object,
    *,
    selectors: _SubjectSelectors,
    reference_kind: str,
) -> list[str]:
    fields = []
    if selectors.references.get(reference_kind) == getattr(row, "id"):
        fields.append("reference")
    if reference_kind == "contact_email" and selectors.email:
        if selectors.email in {
            _normalized_email(getattr(row, "email", None)),
            _normalized_email(getattr(row, "normalized_email", None)),
        }:
            fields.append("email")
    return fields


def _related_actor_match_fields(
    row: object,
    *,
    contact_ids: set[str],
    account_id: str | None,
    actor_fields: Sequence[str],
) -> list[str]:
    fields = []
    if getattr(row, "contact_id", None) in contact_ids:
        fields.append("contact_id")
    fields.extend(_actor_match_fields(row, account_id, actor_fields))
    return fields


def _actor_match_fields(
    row: object,
    account_id: str | None,
    fields: Sequence[str],
) -> list[str]:
    if not account_id:
        return []
    return [field for field in fields if getattr(row, field, None) == account_id]


def _validate_record(record: DsarRecordRef) -> None:
    if record.provider_id != "addresses" or record.module_id != "addresses":
        raise ValueError("Addresses DSAR received a foreign provider record.")


def _validate_action(action: DsarErasureActionRef) -> None:
    if action.provider_id != "addresses" or action.module_id != "addresses":
        raise ValueError("Addresses DSAR received a foreign provider action.")


def _record(
    resource_type: str,
    resource_id: str,
    category: str,
    title: str,
    data: Mapping[str, object],
    *,
    observed_at: datetime | None = None,
    immutable: bool = False,
    retention_reason: str | None = None,
) -> DsarRecordRef:
    return DsarRecordRef(
        provider_id="addresses",
        module_id="addresses",
        resource_type=resource_type,
        resource_id=resource_id,
        category=category,
        title=title,
        data=data,
        observed_at=observed_at,
        immutable_evidence=immutable,
        retention_reason=retention_reason,
        source_path="/address-book",
    )


def _session(value: object) -> Session:
    if not isinstance(value, Session):
        raise TypeError("Addresses DSAR provider requires a SQLAlchemy session.")
    return value


def _bounded_rows(query: object) -> list[object]:
    rows = query.limit(_MAX_RECORDS + 1).all()  # type: ignore[attr-defined]
    if len(rows) > _MAX_RECORDS:
        raise ValueError(
            "Addresses DSAR match limit exceeded; narrow the subject selectors."
        )
    return rows


def _bounded_text(value: str | None, limit: int) -> str | None:
    return value[:limit] if value else None


def _normalized_email(value: object) -> str | None:
    if not isinstance(value, str):
        return None
    value = value.strip().casefold()
    return value or None


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


def _iso(value: datetime | None) -> str | None:
    if value is None:
        return None
    if value.tzinfo is None:
        value = value.replace(tzinfo=timezone.utc)
    return value.isoformat()


__all__ = ["ADDRESSES_DSAR_CAPABILITY", "AddressesDsarProvider"]