leocrm/app/core/audit.py

"""Audit log middleware — records create/update/delete actions."""

from __future__ import annotations

import uuid
from typing import Any

from sqlalchemy.ext.asyncio import AsyncSession

from app.models.audit import AuditLog
from app.models.entity_history import EntityHistory


async def log_audit(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    user_id: uuid.UUID | None,
    action: str,
    entity_type: str,
    entity_id: uuid.UUID | None = None,
    changes: dict[str, Any] | None = None,
    details: dict[str, Any] | None = None,
) -> AuditLog:
    """Create an audit log entry.

    Sensitive fields in *changes* are masked using the central
    :mod:`app.core.sensitive_data` module.
    """
    from app.core.sensitive_data import sanitize_dict

    masked_changes = sanitize_dict(changes, entity_type) if changes else changes
    entry = AuditLog(
        tenant_id=tenant_id,
        user_id=user_id,
        action=action,
        entity_type=entity_type,
        entity_id=entity_id,
        changes=masked_changes,
    )
    db.add(entry)
    await db.flush()
    return entry


async def log_deletion(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    user_id: uuid.UUID | None,
    entity_type: str,
    entity_id: uuid.UUID,
    entity_snapshot: dict[str, Any],
) -> EntityHistory:
    """Create a deletion history entry (merged from DeletionLog into EntityHistory).

    Stores the full entity snapshot in snapshot_before for forensic recovery.
    Sensitive fields are masked using the central
    :mod:`app.core.sensitive_data` module.
    """
    from app.core.sensitive_data import sanitize_dict

    masked_snapshot = sanitize_dict(entity_snapshot, entity_type)
    entry = EntityHistory(
        tenant_id=tenant_id,
        user_id=user_id,
        entity_type=entity_type,
        entity_id=entity_id,
        action="delete",
        snapshot_before=masked_snapshot,
        snapshot_after=None,
        changes=None,
        owner_id=user_id,
    )
    db.add(entry)
    await db.flush()
    return entry