diff --git a/alembic/versions/0098_files_dedup_index.py b/alembic/versions/0098_files_dedup_index.py new file mode 100644 index 0000000..3c0b886 --- /dev/null +++ b/alembic/versions/0098_files_dedup_index.py @@ -0,0 +1,48 @@ +"""Add tenant-local deduplication index on files. + +Plan 6.6: Partial unique index on (tenant_id, content_hash) +WHERE content_hash IS NOT NULL AND deleted_at IS NULL. + +Before creating the unique index, checks for existing duplicates. + +Revision ID: 0098 +Revises: 0097 +""" + +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + +revision = "0098" +down_revision = "0097" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + # Check for duplicates before creating unique index + conn = op.get_bind() + duplicates = conn.execute( + sa.text( + "SELECT tenant_id, content_hash, count(*) FROM files " + "WHERE content_hash IS NOT NULL AND deleted_at IS NULL " + "GROUP BY tenant_id, content_hash HAVING count(*) > 1" + ) + ).fetchall() + + if duplicates: + raise RuntimeError( + f"Cannot create unique index: {len(duplicates)} duplicate (tenant_id, content_hash) pairs found. " + "Data cleanup required before migration." + ) + + op.execute( + "CREATE UNIQUE INDEX IF NOT EXISTS uq_files_tenant_content_hash " + "ON files (tenant_id, content_hash) " + "WHERE content_hash IS NOT NULL AND deleted_at IS NULL" + ) + + +def downgrade() -> None: + op.execute("DROP INDEX IF EXISTS uq_files_tenant_content_hash") diff --git a/app/plugins/builtins/dms/routes.py b/app/plugins/builtins/dms/routes.py index e9bcb95..e3774ce 100644 --- a/app/plugins/builtins/dms/routes.py +++ b/app/plugins/builtins/dms/routes.py @@ -519,19 +519,34 @@ async def upload_file( mime_type = upload_data["mime_type"] - dms_file = DmsFile( - id=file_id, - tenant_id=tenant_id, - name=upload_data["filename"], - folder_id=fid, - uploaded_by=user_id, - mime_type=mime_type, - size_bytes=file_size, - storage_path=storage_path, - content_hash=content_hash, + # Tenant-local deduplication: check for existing file with same content_hash + existing_q = await db.execute( + select(DmsFile).where( + DmsFile.tenant_id == tenant_id, + DmsFile.content_hash == content_hash, + DmsFile.deleted_at.is_(None), + ).limit(1) ) - db.add(dms_file) - await db.flush() + existing_file = existing_q.scalar_one_or_none() + + if existing_file: + # Deduplicate: reuse existing file, remove the duplicate we just saved + await storage.delete(storage_path) + dms_file = existing_file + else: + dms_file = DmsFile( + id=file_id, + tenant_id=tenant_id, + name=upload_data["filename"], + folder_id=fid, + uploaded_by=user_id, + mime_type=mime_type, + size_bytes=file_size, + storage_path=storage_path, + content_hash=content_hash, + ) + db.add(dms_file) + await db.flush() return { "id": str(dms_file.id), @@ -540,7 +555,6 @@ async def upload_file( "uploaded_by": str(dms_file.uploaded_by), "mime_type": dms_file.mime_type, "size_bytes": dms_file.size_bytes, - "content_hash": dms_file.content_hash, "deleted_at": None, "created_at": dms_file.created_at.isoformat() if dms_file.created_at else None, "updated_at": dms_file.updated_at.isoformat() if dms_file.updated_at else None, @@ -580,7 +594,6 @@ async def get_file( "uploaded_by": str(dms_file.uploaded_by), "mime_type": dms_file.mime_type, "size_bytes": dms_file.size_bytes, - "content_hash": dms_file.content_hash, "deleted_at": None, "created_at": dms_file.created_at.isoformat() if dms_file.created_at else None, "updated_at": dms_file.updated_at.isoformat() if dms_file.updated_at else None, @@ -729,7 +742,6 @@ async def update_file( "uploaded_by": str(dms_file.uploaded_by), "mime_type": dms_file.mime_type, "size_bytes": dms_file.size_bytes, - "content_hash": dms_file.content_hash, "deleted_at": None, "created_at": dms_file.created_at.isoformat() if dms_file.created_at else None, "updated_at": dms_file.updated_at.isoformat() if dms_file.updated_at else None, @@ -806,7 +818,6 @@ async def restore_file( "uploaded_by": str(dms_file.uploaded_by), "mime_type": dms_file.mime_type, "size_bytes": dms_file.size_bytes, - "content_hash": dms_file.content_hash, "deleted_at": None, "created_at": dms_file.created_at.isoformat() if dms_file.created_at else None, "updated_at": dms_file.updated_at.isoformat() if dms_file.updated_at else None, @@ -853,16 +864,28 @@ async def preview_file( 404, detail={"detail": "File not found on disk", "code": "file_missing"} ) - content = await storage.read(dms_file.storage_path) + # Stream file directly from storage without loading into RAM + from fastapi.responses import FileResponse as FastApiFileResponse + import os as _os - def _stream(): - yield content - - return StreamingResponse( - _stream(), - media_type="application/pdf", - headers={"Content-Disposition": f'inline; filename="{dms_file.name}"'}, - ) + if isinstance(storage, LocalStorage): + # LocalStorage: use FileResponse for automatic streaming + full_path = storage._full_path(dms_file.storage_path) + return FastApiFileResponse( + path=full_path, + media_type="application/pdf", + headers={"Content-Disposition": f'inline; filename="{dms_file.name}"'}, + ) + else: + # S3 or other: fall back to read (TODO: implement S3 streaming) + content = await storage.read(dms_file.storage_path) + def _stream(): + yield content + return StreamingResponse( + _stream(), + media_type="application/pdf", + headers={"Content-Disposition": f'inline; filename="{dms_file.name}"'}, + ) @router.post("/files/{file_id}/edit-session", dependencies=[Depends(require_permission("dms:write"))]) diff --git a/app/routes/attachments.py b/app/routes/attachments.py index 4e22076..0625a18 100644 --- a/app/routes/attachments.py +++ b/app/routes/attachments.py @@ -34,13 +34,12 @@ async def upload_attachment( except ValueError: raise HTTPException(400, detail={"detail": "Invalid entity_id", "code": "invalid_id"}) from None - file_content = await file.read() mime_type = file.content_type or "application/octet-stream" try: return await attachment_service.save_attachment( db, tenant_id, user_id, entity_type, eid, - file.filename or "unknown", file_content, mime_type, + file.filename or "unknown", file, mime_type, is_system_admin=is_admin, ) except PermissionError as e: diff --git a/app/services/attachment_service.py b/app/services/attachment_service.py index f869a1a..e4d4c41 100644 --- a/app/services/attachment_service.py +++ b/app/services/attachment_service.py @@ -47,8 +47,6 @@ def _entity_attachment_to_dict(ea: EntityAttachment, dms_file: DmsFile | None = "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, @@ -63,20 +61,60 @@ async def save_attachment( entity_type: str, entity_id: uuid.UUID, filename: str, - file_content: bytes, + file: Any, # UploadFile or async iterator of chunks 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})") + """Save a file to DMS and create an entity_attachments reference. + + Streams the file in chunks to avoid loading entire file into RAM. + """ + import hashlib + from app.core.storage import get_storage_backend + + # Generate unique filename and storage path + unique_filename = _generate_unique_filename(filename) + storage_path = f"attachments/{entity_type}/{entity_id}/{unique_filename}" + + # Stream file to storage — compute hash and size during streaming + sha256 = hashlib.sha256() + file_size = 0 + CHUNK_SIZE = 1024 * 1024 # 1MB chunks + + async def chunk_stream(): + nonlocal file_size + if hasattr(file, 'read'): + # UploadFile object + while True: + chunk = await file.read(CHUNK_SIZE) + if not chunk: + break + file_size += len(chunk) + sha256.update(chunk) + yield chunk + else: + # Already bytes (backward compat) + nonlocal_bytes = file if isinstance(file, bytes) else b''.join([c async for c in file]) + file_size = len(nonlocal_bytes) + sha256.update(nonlocal_bytes) + yield nonlocal_bytes + + # Check file size limit during streaming + # (we check after streaming — for true streaming we'd need a wrapper) + # For now, stream and check size after + storage = get_storage_backend() + await storage.save_stream(storage_path, chunk_stream()) + + if file_size > MAX_FILE_SIZE: + await storage.delete(storage_path) + raise ValueError(f"File too large: {file_size} 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: + await storage.delete(storage_path) raise ValueError(f"File type not allowed: {_ext}") # Check access on parent entity @@ -85,14 +123,10 @@ async def save_attachment( db, entity_type, entity_id, user_id, tenant_id, "write", is_system_admin ) if not has_access: + await storage.delete(storage_path) 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() + content_hash = sha256.hexdigest() # Check for existing DMS file with same hash in same tenant (deduplication) existing_file = await db.execute( @@ -107,19 +141,16 @@ async def save_attachment( if existing_dms_file: # Deduplicate: reuse existing DMS file, just create new reference dms_file = existing_dms_file + await storage.delete(storage_path) # Remove the duplicate we just saved else: - # Save file via storage backend - storage = get_storage_backend() - await storage.save(storage_path, file_content) - - # Create DMS File record + # File already streamed to storage — 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), + size_bytes=file_size, storage_path=storage_path, content_hash=content_hash, owner_id=user_id,