leocrm/app/services/entity_history_service.py

"""Entity history service — record, query, restore, and undo entity snapshots."""

from __future__ import annotations

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

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

from app.core.visibility import apply_visibility_filter, check_single_entity_access
from app.models.entity_history import EntityHistory


async def record_history(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    user_id: uuid.UUID | None,
    entity_type: str,
    entity_id: uuid.UUID,
    action: str,
    snapshot_before: dict[str, Any] | None = None,
    snapshot_after: dict[str, Any] | None = None,
    changes: dict[str, Any] | None = None,
) -> EntityHistory:
    """Create a history entry for a CRUD action."""
    entry = EntityHistory(
        tenant_id=tenant_id,
        user_id=user_id,
        entity_type=entity_type,
        entity_id=entity_id,
        action=action,
        snapshot_before=snapshot_before,
        snapshot_after=snapshot_after,
        changes=changes,
        owner_id=user_id,
    )
    db.add(entry)
    await db.flush()
    return entry


async def get_entity_history(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    entity_type: str,
    entity_id: uuid.UUID,
    limit: int = 50,
    user_id: uuid.UUID | None = None,
    is_system_admin: bool = False,
) -> list[EntityHistory]:
    """Get all history entries for an entity, newest first."""
    q = (
        select(EntityHistory)
        .where(
            EntityHistory.tenant_id == tenant_id,
            EntityHistory.entity_type == entity_type,
            EntityHistory.entity_id == entity_id,
        )
        .order_by(EntityHistory.created_at.desc())
        .limit(limit)
    )
    if user_id and not is_system_admin:
        q = await apply_visibility_filter(
            db, q, "entity_history", EntityHistory, user_id, tenant_id, is_system_admin
        )
    result = await db.execute(q)
    return list(result.scalars().all())


async def get_history_entry(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    history_id: uuid.UUID,
    user_id: uuid.UUID | None = None,
    is_system_admin: bool = False,
) -> EntityHistory | None:
    """Get a specific history entry by ID."""
    q = select(EntityHistory).where(
        EntityHistory.id == history_id,
        EntityHistory.tenant_id == tenant_id,
    )
    result = await db.execute(q)
    entry = result.scalar_one_or_none()
    if entry is None:
        return None
    if user_id and not is_system_admin:
        has_access = await check_single_entity_access(
            db, "entity_history", entry.id, user_id, tenant_id, "read", is_system_admin
        )
        if not has_access:
            raise PermissionError("No access")
    return entry


async def restore_from_history(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    history_id: uuid.UUID,
    user_id: uuid.UUID,
    is_system_admin: bool = False,
) -> dict[str, Any]:
    """Restore an entity to a previous snapshot state.

    For 'delete' actions: un-delete the entity (clear deleted_at).
    For 'update' actions: revert entity fields to snapshot_before.
    For 'create' actions: soft-delete the entity (undo creation).

    Returns the restored data dict.
    """
    entry = await get_history_entry(db, tenant_id, history_id, user_id, is_system_admin)
    if entry is None:
        raise ValueError("History entry not found")

    entity_type = entry.entity_type
    entity_id = entry.entity_id
    action = entry.action

    # Import here to avoid circular imports
    from app.models.contact import Contact

    if entity_type == "contact":
        q = select(Contact).where(
            Contact.id == entity_id,
            Contact.tenant_id == tenant_id,
        )
        result = await db.execute(q)
        contact = result.scalar_one_or_none()

        if action == "delete":
            # Un-delete: clear deleted_at
            if contact is None:
                raise ValueError("Entity not found for restore")
            contact.deleted_at = None
            contact.updated_by = user_id
            await db.flush()
            from app.services.contact_service import _serialize_contact_detail
            from sqlalchemy.orm import selectinload
            q2 = (
                select(Contact)
                .options(selectinload(Contact.contact_persons))
                .where(Contact.id == contact.id)
            )
            result2 = await db.execute(q2)
            contact = result2.scalar_one()
            return _serialize_contact_detail(contact)

        elif action == "update":
            # Revert to snapshot_before
            if contact is None:
                raise ValueError("Entity not found for restore")
            if entry.snapshot_before is None:
                raise ValueError("No snapshot_before available for restore")
            for key, value in entry.snapshot_before.items():
                if hasattr(contact, key) and key not in ("id", "tenant_id", "created_at", "updated_at", "deleted_at"):
                    setattr(contact, key, value)
            contact.updated_by = user_id
            await db.flush()
            from app.services.contact_service import _serialize_contact_detail
            from sqlalchemy.orm import selectinload
            q2 = (
                select(Contact)
                .options(selectinload(Contact.contact_persons))
                .where(Contact.id == contact.id)
            )
            result2 = await db.execute(q2)
            contact = result2.scalar_one()
            return _serialize_contact_detail(contact)

        elif action == "create":
            # Undo creation: soft-delete the entity
            if contact is None:
                raise ValueError("Entity not found for restore")
            contact.deleted_at = datetime.now(timezone.utc)
            contact.updated_by = user_id
            await db.flush()
            from app.services.contact_service import _serialize_contact_detail
            from sqlalchemy.orm import selectinload
            q2 = (
                select(Contact)
                .options(selectinload(Contact.contact_persons))
                .where(Contact.id == contact.id)
            )
            result2 = await db.execute(q2)
            contact = result2.scalar_one()
            return _serialize_contact_detail(contact)

    raise ValueError(f"Unsupported entity type for restore: {entity_type}")


async def undo_last_action(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    user_id: uuid.UUID,
    entity_type: str,
    entity_id: uuid.UUID,
    is_system_admin: bool = False,
) -> dict[str, Any]:
    """Undo the most recent action for an entity.

    Returns the restored entity data.
    Raises ValueError if no history exists.
    """
    q = (
        select(EntityHistory)
        .where(
            EntityHistory.tenant_id == tenant_id,
            EntityHistory.entity_type == entity_type,
            EntityHistory.entity_id == entity_id,
        )
        .order_by(EntityHistory.created_at.desc())
        .limit(1)
    )
    if not is_system_admin:
        q = await apply_visibility_filter(
            db, q, "entity_history", EntityHistory, user_id, tenant_id, is_system_admin
        )
    result = await db.execute(q)
    entry = result.scalar_one_or_none()
    if entry is None:
        raise ValueError("No history found for this entity")

    return await restore_from_history(db, tenant_id, entry.id, user_id, is_system_admin)