leocrm/app/plugins/builtins/contacts/contracts.py

"""Public contract for the contacts plugin.

Exposes the symbols that other core modules and plugins need without
importing from internal modules directly (Block C7: dashboard counts).
"""

from __future__ import annotations

from typing import Any

from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.visibility import apply_visibility_filter
from app.models.contact import Contact
from app.plugins.builtins.contracts import get_contract_registry


class ContactsContract:
    """Public API surface for the contacts plugin."""

    contract_name = "contacts"

    @staticmethod
    async def get_counts(
        db: AsyncSession,
        tenant_id: Any,
        user_id: Any,
        is_system_admin: bool = False,
    ) -> dict[str, int]:
        """Return visibility-filtered contact/company/person counts."""
        queries = []
        for type_filter in (None, "company", "person"):
            query = select(func.count(Contact.id)).where(
                Contact.tenant_id == tenant_id,
                Contact.deleted_at.is_(None),
            )
            if type_filter is not None:
                query = query.where(Contact.type == type_filter)
            query = await apply_visibility_filter(
                db, query, "contact", Contact, user_id, tenant_id, is_system_admin
            )
            queries.append(query)

        results = [((await db.execute(q)).scalar() or 0) for q in queries]
        return {
            "contacts": results[0],
            "companies": results[1],
            "persons": results[2],
            "total": results[0],
        }

    @staticmethod
    async def dsar_collect(
        db: AsyncSession, tenant_id: Any, user_id: Any
    ) -> dict[str, Any]:
        """GDPR Art. 15: collect contact data owned by the user.

        Owned by the contacts plugin — core/jobs.py calls this generically
        via the contract, it must not know contact internals.
        """
        contacts = (
            await db.execute(
                select(Contact).where(
                    Contact.tenant_id == tenant_id,
                    Contact.owner_id == user_id,
                    Contact.deleted_at.is_(None),
                )
            )
        ).scalars().all()
        return {
            "contacts": [
                {
                    "id": str(c.id),
                    "type": c.type,
                    "displayname": c.displayname,
                    "email_1": c.email_1,
                    "email_2": c.email_2,
                }
                for c in contacts
            ]
        }

    @staticmethod
    async def dsar_erase(
        db: AsyncSession, tenant_id: Any, user_id: Any
    ) -> dict[str, int]:
        """GDPR Art. 17: soft-delete contacts owned by the user.

        Soft-delete via deleted_at (audit history must remain intact — it is
        a business record, not personal data of the subject; retention
        policy governs its cleanup).
        """
        from datetime import UTC, datetime

        contacts = (
            await db.execute(
                select(Contact).where(
                    Contact.tenant_id == tenant_id,
                    Contact.owner_id == user_id,
                    Contact.deleted_at.is_(None),
                )
            )
        ).scalars().all()
        for c in contacts:
            c.deleted_at = datetime.now(UTC)
        return {"contacts_soft_deleted": len(contacts)}

    @classmethod
    def get_function(cls, name: str):
        """Return a callable exposed by this contract, or None if absent."""
        return getattr(cls, name, None)


# ─── self-registration ───

_contract = ContactsContract()
get_contract_registry().register("contacts", _contract)