leocrm/app/services/attachment_service.py

"""Attachment service — unified through DMS.

All files are stored in the DMS (files table).
entity_attachments just references the DMS file with entity_type/entity_id.

This unifies: one upload path, one download path, one permission model,
one deduplication (content_hash), one storage backend.
"""

from __future__ import annotations

import hashlib
import os
import uuid
from datetime import UTC, datetime
from typing import Any

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

from app.core.audit import log_audit
from app.core.storage import get_storage_backend
from app.core.visibility import apply_visibility_filter, check_single_entity_access
from app.models.entity_attachment import EntityAttachment
from app.plugins.builtins.dms.models import File as DmsFile


# File size limit: 50MB
MAX_FILE_SIZE = 50 * 1024 * 1024


def _generate_unique_filename(original_filename: str) -> str:
    """Generate a unique filename using UUID + original extension."""
    ext = os.path.splitext(original_filename)[1]
    return f"{uuid.uuid4().hex}{ext}"


def _entity_attachment_to_dict(ea: EntityAttachment, dms_file: DmsFile | None = None) -> dict[str, Any]:
    """Serialize an EntityAttachment + DMS File to dict."""
    return {
        "id": str(ea.id),
        "entity_type": ea.entity_type,
        "entity_id": str(ea.entity_id),
        "dms_file_id": str(ea.dms_file_id),
        "category": ea.category,
        "display_name": ea.display_name,
        "filename": dms_file.name if dms_file else (ea.display_name or "unknown"),
        "mime_type": dms_file.mime_type if dms_file else "application/octet-stream",
        "file_size": dms_file.size_bytes if dms_file else 0,
        "storage_path": dms_file.storage_path if dms_file else None,
        "content_hash": dms_file.content_hash if dms_file else None,
        "uploaded_by": str(ea.created_by) if ea.created_by else None,
        "owner_id": str(ea.owner_id) if ea.owner_id else None,
        "created_at": ea.created_at.isoformat() if ea.created_at else None,
        "updated_at": ea.updated_at.isoformat() if ea.updated_at else None,
    }


async def save_attachment(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    user_id: uuid.UUID,
    entity_type: str,
    entity_id: uuid.UUID,
    filename: str,
    file_content: bytes,
    mime_type: str,
    is_system_admin: bool = False,
) -> dict[str, Any]:
    """Save a file to DMS and create an entity_attachments reference."""
    # File size limit
    if len(file_content) > MAX_FILE_SIZE:
        raise ValueError(f"File too large: {len(file_content)} bytes (max {MAX_FILE_SIZE})")

    # Check for blocked file types
    import os as _os
    _BLOCKED = {".exe", ".bat", ".cmd", ".sh", ".jar", ".com", ".scr", ".msi", ".dll", ".vbs", ".ps1", ".app", ".bin", ".reg", ".inf"}
    _ext = _os.path.splitext(filename)[1].lower()
    if _ext in _BLOCKED:
        raise ValueError(f"File type not allowed: {_ext}")

    # Check access on parent entity
    if not is_system_admin:
        has_access = await check_single_entity_access(
            db, entity_type, entity_id, user_id, tenant_id, "write", is_system_admin
        )
        if not has_access:
            raise PermissionError(f"No write access to {entity_type} {entity_id}")

    # Generate unique filename and storage path
    unique_filename = _generate_unique_filename(filename)
    storage_path = f"attachments/{entity_type}/{entity_id}/{unique_filename}"

    # Calculate content hash for deduplication (tenant-local)
    content_hash = hashlib.sha256(file_content).hexdigest()

    # Check for existing DMS file with same hash in same tenant (deduplication)
    existing_file = await db.execute(
        select(DmsFile).where(
            DmsFile.tenant_id == tenant_id,
            DmsFile.content_hash == content_hash,
            DmsFile.deleted_at.is_(None),
        ).limit(1)
    )
    existing_dms_file = existing_file.scalar_one_or_none()

    if existing_dms_file:
        # Deduplicate: reuse existing DMS file, just create new reference
        dms_file = existing_dms_file
    else:
        # Save file via storage backend
        storage = get_storage_backend()
        await storage.save(storage_path, file_content)

        # Create DMS File record
        dms_file = DmsFile(
            tenant_id=tenant_id,
            name=filename,
            folder_id=None,  # Attachments don't go in DMS folders
            uploaded_by=user_id,
            mime_type=mime_type,
            size_bytes=len(file_content),
            storage_path=storage_path,
            content_hash=content_hash,
            owner_id=user_id,
        )
        db.add(dms_file)
        await db.flush()
        await db.refresh(dms_file)

    # Create entity_attachments reference
    entity_attachment = EntityAttachment(
        tenant_id=tenant_id,
        entity_type=entity_type,
        entity_id=entity_id,
        dms_file_id=dms_file.id,
        category=None,
        display_name=filename,
        owner_id=user_id,
        created_by=user_id,
    )
    db.add(entity_attachment)
    await db.flush()
    await db.refresh(entity_attachment)

    await log_audit(
        db, tenant_id, user_id, "upload", "attachment", entity_attachment.id,
        changes={"filename": filename, "entity_type": entity_type, "entity_id": str(entity_id), "dms_file_id": str(dms_file.id)},
    )
    return _entity_attachment_to_dict(entity_attachment, dms_file)


async def list_attachments(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    entity_type: str,
    entity_id: uuid.UUID,
    user_id: uuid.UUID | None = None,
    is_system_admin: bool = False,
) -> dict[str, Any]:
    """List attachments for a specific entity (via DMS files)."""
    q = (
        select(EntityAttachment, DmsFile)
        .join(DmsFile, EntityAttachment.dms_file_id == DmsFile.id)
        .where(
            EntityAttachment.tenant_id == tenant_id,
            EntityAttachment.entity_type == entity_type,
            EntityAttachment.entity_id == entity_id,
            EntityAttachment.deleted_at.is_(None),
            DmsFile.deleted_at.is_(None),
        )
        .order_by(EntityAttachment.created_at.desc())
    )
    if user_id and not is_system_admin:
        q = await apply_visibility_filter(
            db, q, "entity_attachment", EntityAttachment, user_id, tenant_id, is_system_admin
        )
    result = await db.execute(q)
    rows = result.all()
    return {
        "items": [_entity_attachment_to_dict(ea, df) for ea, df in rows],
        "total": len(rows),
    }


async def get_attachment(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    attachment_id: uuid.UUID,
    user_id: uuid.UUID | None = None,
    is_system_admin: bool = False,
) -> dict[str, Any] | None:
    """Get a single attachment by ID (with DMS file info)."""
    q = (
        select(EntityAttachment, DmsFile)
        .join(DmsFile, EntityAttachment.dms_file_id == DmsFile.id)
        .where(
            EntityAttachment.id == attachment_id,
            EntityAttachment.tenant_id == tenant_id,
            EntityAttachment.deleted_at.is_(None),
        )
    )
    result = await db.execute(q)
    row = result.first()
    if row is None:
        return None
    ea, dms_file = row
    if user_id and not is_system_admin:
        has_access = await check_single_entity_access(
            db, "entity_attachment", ea.id, user_id, tenant_id, "read", is_system_admin
        )
        if not has_access:
            raise PermissionError("No access")
    return _entity_attachment_to_dict(ea, dms_file)


async def get_attachment_download_path(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    attachment_id: uuid.UUID,
    user_id: uuid.UUID | None = None,
    is_system_admin: bool = False,
) -> str | None:
    """Get the storage path for downloading an attachment's DMS file."""
    q = (
        select(EntityAttachment, DmsFile)
        .join(DmsFile, EntityAttachment.dms_file_id == DmsFile.id)
        .where(
            EntityAttachment.id == attachment_id,
            EntityAttachment.tenant_id == tenant_id,
            EntityAttachment.deleted_at.is_(None),
            DmsFile.deleted_at.is_(None),
        )
    )
    result = await db.execute(q)
    row = result.first()
    if row is None:
        return None
    ea, dms_file = row
    if user_id and not is_system_admin:
        has_access = await check_single_entity_access(
            db, "entity_attachment", ea.id, user_id, tenant_id, "read", is_system_admin
        )
        if not has_access:
            raise PermissionError("No access")
    return dms_file.storage_path


async def delete_attachment(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    user_id: uuid.UUID,
    attachment_id: uuid.UUID,
    is_system_admin: bool = False,
) -> bool:
    """Soft-delete an entity_attachments reference.
    
    The DMS file is NOT deleted because other entities may reference it.
    DMS file cleanup happens via DMS's own deletion workflow.
    """
    q = select(EntityAttachment).where(
        EntityAttachment.id == attachment_id,
        EntityAttachment.tenant_id == tenant_id,
        EntityAttachment.deleted_at.is_(None),
    )
    result = await db.execute(q)
    ea = result.scalar_one_or_none()
    if ea is None:
        return False

    if not is_system_admin:
        has_access = await check_single_entity_access(
            db, "entity_attachment", ea.id, user_id, tenant_id, "admin", is_system_admin
        )
        if not has_access:
            raise PermissionError("No access")

    ea.deleted_at = datetime.now(UTC)
    await db.flush()
    await log_audit(
        db, tenant_id, user_id, "delete", "attachment", attachment_id,
        changes={"display_name": ea.display_name, "dms_file_id": str(ea.dms_file_id)},
    )
    return True