erp-nutzfahrzeuge/backend/app/services/contact_service.py

"""Contact service: CRUD, search, filter, pagination, and soft-delete operations."""

from __future__ import annotations

import uuid
from datetime import datetime, timezone
from typing import Any

from sqlalchemy import and_, func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload

from app.models.contact import Contact, ContactPerson


_SORTABLE_FIELDS: set[str] = {
    "company_name",
    "address_city",
    "address_country",
    "role",
    "created_at",
    "updated_at",
    "email",
}


def _apply_filters(
    stmt: select,
    search: str | None = None,
    role: str | None = None,
    is_eu: bool | None = None,
    is_private: bool | None = None,
) -> select:
    """Apply WHERE filters to a select statement (always excludes soft-deleted)."""
    conditions = [Contact.deleted_at.is_(None)]

    if search:
        search_pattern = f"%{search}%"
        conditions.append(
            or_(
                Contact.company_name.ilike(search_pattern),
                Contact.address_city.ilike(search_pattern),
                Contact.email.ilike(search_pattern),
                Contact.vat_id.ilike(search_pattern),
            )
        )

    if role:
        # When filtering by 'kaeufer' or 'verkaeufer', include 'beide' as well
        if role in ("kaeufer", "verkaeufer"):
            conditions.append(
                or_(
                    Contact.role == role,
                    Contact.role == "beide",
                )
            )
        else:
            conditions.append(Contact.role == role)

    if is_eu is not None:
        if is_eu:
            # EU = address_country != 'DE'
            conditions.append(Contact.address_country != "DE")
        else:
            # Inland = address_country == 'DE'
            conditions.append(Contact.address_country == "DE")

    if is_private is not None:
        conditions.append(Contact.is_private == is_private)

    return stmt.where(and_(*conditions))


def _apply_sort(stmt: select, sort: str | None = None) -> select:
    """Apply ORDER BY to a select statement based on sort param.

    Format: 'field' for ascending, '-field' for descending.
    """
    if not sort:
        return stmt.order_by(Contact.created_at.desc())

    descending = sort.startswith("-")
    field_name = sort.lstrip("-")

    if field_name not in _SORTABLE_FIELDS:
        return stmt.order_by(Contact.created_at.desc())

    column = getattr(Contact, field_name)
    if descending:
        return stmt.order_by(column.desc())
    return stmt.order_by(column.asc())


async def list_contacts(
    db: AsyncSession,
    page: int = 1,
    page_size: int = 20,
    search: str | None = None,
    role: str | None = None,
    is_eu: bool | None = None,
    is_private: bool | None = None,
    sort: str | None = None,
) -> tuple[list[Contact], int]:
    """List contacts with pagination, filtering, and sorting.

    Returns (contacts, total_count).
    """
    # Build count query
    count_stmt = select(func.count(Contact.id))
    count_stmt = _apply_filters(
        count_stmt,
        search=search,
        role=role,
        is_eu=is_eu,
        is_private=is_private,
    )
    total_result = await db.execute(count_stmt)
    total = total_result.scalar_one()

    # Build data query with eager-loaded contact persons
    data_stmt = select(Contact).options(selectinload(Contact.contact_persons))
    data_stmt = _apply_filters(
        data_stmt,
        search=search,
        role=role,
        is_eu=is_eu,
        is_private=is_private,
    )
    data_stmt = _apply_sort(data_stmt, sort)

    offset = (page - 1) * page_size
    data_stmt = data_stmt.offset(offset).limit(page_size)

    result = await db.execute(data_stmt)
    contacts = list(result.scalars().all())

    return contacts, total


async def get_contact_by_id(db: AsyncSession, contact_id: uuid.UUID) -> Contact | None:
    """Get a single contact by ID, excluding soft-deleted. Eager-loads contact persons."""
    stmt = (
        select(Contact)
        .options(selectinload(Contact.contact_persons))
        .where(and_(Contact.id == contact_id, Contact.deleted_at.is_(None)))
    )
    result = await db.execute(stmt)
    return result.scalar_one_or_none()


async def create_contact(db: AsyncSession, data: dict[str, Any]) -> Contact:
    """Create a new contact with optional nested contact persons.

    The data dict may contain a 'contact_persons' list of dicts.
    """
    persons_data = data.pop("contact_persons", [])

    contact = Contact(**data)
    db.add(contact)
    await db.flush()  # Flush to generate contact.id

    # Add contact persons if provided
    for person_data in persons_data:
        person = ContactPerson(
            contact_id=contact.id,
            name=person_data["name"],
            function=person_data.get("function"),
            phone=person_data.get("phone"),
            email=person_data.get("email"),
        )
        db.add(person)

    await db.flush()
    await db.refresh(contact)
    return contact


async def update_contact(
    db: AsyncSession, contact_id: uuid.UUID, updates: dict[str, Any]
) -> Contact | None:
    """Update a contact's fields. Returns None if not found or deleted."""
    contact = await get_contact_by_id(db, contact_id)
    if contact is None:
        return None

    for key, value in updates.items():
        if hasattr(contact, key):
            setattr(contact, key, value)

    await db.flush()
    await db.refresh(contact)
    return contact


async def soft_delete_contact(
    db: AsyncSession, contact_id: uuid.UUID
) -> Contact | None:
    """Soft-delete a contact by setting deleted_at. Returns None if not found."""
    contact = await get_contact_by_id(db, contact_id)
    if contact is None:
        return None

    contact.deleted_at = datetime.now(timezone.utc)
    await db.flush()
    await db.refresh(contact)
    return contact


async def add_contact_person(
    db: AsyncSession,
    contact_id: uuid.UUID,
    person_data: dict[str, Any],
) -> ContactPerson | None:
    """Add a contact person to an existing contact.

    Returns None if the contact is not found or is deleted.
    """
    contact = await get_contact_by_id(db, contact_id)
    if contact is None:
        return None

    person = ContactPerson(
        contact_id=contact_id,
        name=person_data["name"],
        function=person_data.get("function"),
        phone=person_data.get("phone"),
        email=person_data.get("email"),
    )
    db.add(person)
    await db.flush()
    await db.refresh(person)
    return person


async def remove_contact_person(
    db: AsyncSession,
    contact_id: uuid.UUID,
    person_id: uuid.UUID,
) -> bool:
    """Remove (hard-delete) a contact person from a contact.

    Returns True if the person was found and deleted, False otherwise.
    """
    stmt = select(ContactPerson).where(
        and_(
            ContactPerson.id == person_id,
            ContactPerson.contact_id == contact_id,
        )
    )
    result = await db.execute(stmt)
    person = result.scalar_one_or_none()
    if person is None:
        return False

    await db.delete(person)
    await db.flush()
    return True