gh-christianlouis-docuelevate/app/api/files.py

"""
File-related API endpoints
"""

import io
import logging
import mimetypes
import os
import uuid
import zipfile
from datetime import datetime, timezone
from typing import Annotated, List, Optional

import aiofiles
from fastapi import APIRouter, Depends, File, HTTPException, Query, Request, UploadFile, status
from fastapi.responses import StreamingResponse
from sqlalchemy import asc, desc
from sqlalchemy.orm import Session

from app.auth import require_login
from app.config import settings
from app.database import get_db
from app.middleware.upload_rate_limit import require_upload_rate_limit
from app.models import FileProcessingStep, FileRecord, ProcessingLog
from app.tasks.convert_to_pdf import convert_to_pdf
from app.tasks.process_document import process_document
from app.utils.allowed_types import ALLOWED_EXTENSIONS, ALLOWED_MIME_TYPES, IMAGE_MIME_TYPES
from app.utils.file_operations import hash_file
from app.utils.file_queries import apply_status_filter
from app.utils.file_status import get_files_processing_status
from app.utils.filename_utils import sanitize_filename
from app.utils.input_validation import validate_search_query, validate_sort_field, validate_sort_order
from app.utils.user_scope import apply_owner_filter, get_current_owner_id

# Set up logging
logger = logging.getLogger(__name__)

router = APIRouter()

DbSession = Annotated[Session, Depends(get_db)]


def get_limiter():
    """Get the limiter from app state."""
    from app.main import app

    return app.state.limiter


@router.get("/files")
@require_login
def list_files_api(
    request: Request,
    db: DbSession,
    page: int = Query(1, ge=1, description="Page number"),
    per_page: int = Query(25, ge=1, le=200, description="Items per page"),
    sort_by: str = Query(
        "created_at",
        description="Sort field: id, original_filename, file_size, mime_type, created_at, status",
    ),
    sort_order: str = Query("desc", description="Sort order: asc or desc"),
    search: Optional[str] = Query(None, description="Search in filename"),
    mime_type: Optional[str] = Query(None, description="Filter by MIME type"),
    status: Optional[str] = Query(None, description="Filter by processing status"),
    date_from: Optional[str] = Query(None, description="Filter files created on or after this date (ISO 8601)"),
    date_to: Optional[str] = Query(None, description="Filter files created on or before this date (ISO 8601)"),
    storage_provider: Optional[str] = Query(None, description="Filter by storage provider (e.g. dropbox, s3)"),
    tags: Optional[str] = Query(None, description="Filter by tag (comma-separated for multiple, AND logic)"),
):
    """
    Returns a paginated JSON list of FileRecord entries with processing status.
    Supports server-side sorting, filtering, and searching.

    Query Parameters:
    - page: Page number (default: 1)
    - per_page: Items per page (default: 25, max: 200)
    - sort_by: Field to sort by (default: created_at)
    - sort_order: asc or desc (default: desc)
    - search: Search in filename
    - mime_type: Filter by MIME type
    - status: Filter by processing status (pending, processing, completed, failed)
    - date_from: Filter files created on or after this date (ISO 8601, e.g. 2026-01-01)
    - date_to: Filter files created on or before this date (ISO 8601, e.g. 2026-12-31)
    - storage_provider: Filter by storage provider (e.g. dropbox, s3, google_drive)
    - tags: Filter by tags (comma-separated, AND logic)

    Example response:
    {
      "files": [...],
      "pagination": {
        "page": 1,
        "per_page": 25,
        "total": 150,
        "pages": 6,
        "next": "http://host/api/files?page=2",
        "previous": null
      }
    }
    """
    # Validate and sanitize query parameters
    validate_sort_field(sort_by)
    validate_sort_order(sort_order)
    search = validate_search_query(search)

    # Start with base query, scoped to the current user in multi-user mode
    query = db.query(FileRecord)
    query = apply_owner_filter(query, request)

    # Apply search filter
    if search:
        query = query.filter(FileRecord.original_filename.ilike(f"%{search}%"))

    # Apply MIME type filter
    if mime_type:
        query = query.filter(FileRecord.mime_type == mime_type)

    # Apply date range filters
    if date_from:
        try:
            dt_from = datetime.fromisoformat(date_from).replace(tzinfo=timezone.utc)
            query = query.filter(FileRecord.created_at >= dt_from)
        except ValueError:
            raise HTTPException(
                status_code=422,
                detail="Invalid date_from format. Use ISO 8601 (e.g. 2026-01-01)",
            )

    if date_to:
        try:
            dt_to = datetime.fromisoformat(date_to).replace(tzinfo=timezone.utc)
            query = query.filter(FileRecord.created_at <= dt_to)
        except ValueError:
            raise HTTPException(
                status_code=422,
                detail="Invalid date_to format. Use ISO 8601 (e.g. 2026-12-31)",
            )

    # Apply storage provider filter (files that have a successful upload_to_{provider} step)
    if storage_provider:
        step_name = f"upload_to_{storage_provider}"
        uploaded_file_ids = (
            db.query(FileProcessingStep.file_id)
            .filter(
                FileProcessingStep.step_name == step_name,
                FileProcessingStep.status == "success",
            )
            .distinct()
            .subquery()
        )
        query = query.filter(FileRecord.id.in_(db.query(uploaded_file_ids.c.file_id)))

    # Apply tags filter (AND logic: all specified tags must be present in ai_metadata)
    if tags:
        tag_list = [t.strip().lower() for t in tags.split(",") if t.strip()]
        for tag in tag_list:
            # Escape SQL LIKE wildcards to prevent unintended pattern matching
            escaped_tag = tag.replace("%", r"\%").replace("_", r"\_")
            query = query.filter(FileRecord.ai_metadata.ilike(f"%{escaped_tag}%"))

    # Apply status filter (before pagination for correct counts)
    query = apply_status_filter(query, db, status)

    # Get total count before pagination (after all filters)
    total_items = query.count()

    # Apply sorting
    sort_column = {
        "id": FileRecord.id,
        "original_filename": FileRecord.original_filename,
        "file_size": FileRecord.file_size,
        "mime_type": FileRecord.mime_type,
        "created_at": FileRecord.created_at,
    }.get(sort_by, FileRecord.created_at)

    if sort_order == "asc":
        query = query.order_by(asc(sort_column))
    else:
        query = query.order_by(desc(sort_column))

    # Apply pagination
    offset = (page - 1) * per_page
    files = query.offset(offset).limit(per_page).all()

    # Get processing status for all files efficiently
    file_ids = [f.id for f in files]
    statuses = get_files_processing_status(db, file_ids)

    # Build result with processing status
    result = []
    for f in files:
        result.append(
            {
                "id": f.id,
                "filehash": f.filehash,
                "original_filename": f.original_filename,
                "local_filename": f.local_filename,
                "file_size": f.file_size,
                "mime_type": f.mime_type,
                "created_at": f.created_at.isoformat() if f.created_at else None,
                "processing_status": statuses.get(
                    f.id,
                    {
                        "status": "pending",
                        "last_step": None,
                        "has_errors": False,
                        "total_steps": 0,
                    },
                ),
            }
        )

    # Calculate pagination info
    total_pages = (total_items + per_page - 1) // per_page

    # Build next / previous page URLs by replacing the page query parameter
    next_url = str(request.url.include_query_params(page=page + 1)) if page < total_pages else None
    previous_url = str(request.url.include_query_params(page=page - 1)) if page > 1 else None

    return {
        "files": result,
        "pagination": {
            "page": page,
            "per_page": per_page,
            "total": total_items,
            "pages": total_pages,
            "next": next_url,
            "previous": previous_url,
        },
    }


def _get_file_processing_status(db: Session, file_id: int) -> dict:
    """
    Deprecated: Use app.utils.file_status.get_file_processing_status instead.
    Kept for backward compatibility with file detail endpoint.
    """
    from app.utils.file_status import get_file_processing_status

    return get_file_processing_status(db, file_id)


@router.get("/files/{file_id}")
@require_login
def get_file_details(request: Request, file_id: int, db: DbSession):
    """
    Get detailed information about a specific file including processing history.
    """
    # Find the file record, scoped to the current user in multi-user mode
    query = db.query(FileRecord).filter(FileRecord.id == file_id)
    query = apply_owner_filter(query, request)
    file_record = query.first()

    if not file_record:
        raise HTTPException(status_code=404, detail=f"File record with ID {file_id} not found")

    # Get processing logs
    logs = (
        db.query(ProcessingLog).filter(ProcessingLog.file_id == file_id).order_by(ProcessingLog.timestamp.desc()).all()
    )

    # Build log list
    log_list = []
    for log in logs:
        log_list.append(
            {
                "id": log.id,
                "task_id": log.task_id,
                "step_name": log.step_name,
                "status": log.status,
                "message": log.message,
                "timestamp": log.timestamp.isoformat() if log.timestamp else None,
            }
        )

    # Get processing status
    processing_status = _get_file_processing_status(db, file_id)

    # Check if files exist on disk
    files_on_disk = {"original": (os.path.exists(file_record.local_filename) if file_record.local_filename else False)}

    return {
        "file": {
            "id": file_record.id,
            "filehash": file_record.filehash,
            "original_filename": file_record.original_filename,
            "local_filename": file_record.local_filename,
            "file_size": file_record.file_size,
            "mime_type": file_record.mime_type,
            "created_at": (file_record.created_at.isoformat() if file_record.created_at else None),
        },
        "processing_status": processing_status,
        "logs": log_list,
        "files_on_disk": files_on_disk,
    }


@router.delete("/files/{file_id}")
@require_login
def delete_file_record(request: Request, file_id: int, db: DbSession):
    """
    Delete a file record from the database.
    This only removes the database entry, not the actual file.
    """
    # Check if file deletion is allowed
    if not settings.allow_file_delete:
        raise HTTPException(status_code=403, detail="File deletion is disabled in the configuration")

    try:
        # Find the file record, scoped to the current user in multi-user mode
        query = db.query(FileRecord).filter(FileRecord.id == file_id)
        query = apply_owner_filter(query, request)
        file_record = query.first()

        if not file_record:
            raise HTTPException(status_code=404, detail=f"File record with ID {file_id} not found")

        # Log the deletion
        logger.info(f"Deleting file record: ID={file_id}, Filename={file_record.original_filename}")

        # Delete the record
        db.delete(file_record)
        db.commit()

        return {
            "status": "success",
            "message": f"File record {file_id} deleted successfully",
        }

    except HTTPException:
        raise
    except Exception as e:
        db.rollback()
        logger.exception(f"Error deleting file record {file_id}: {str(e)}")
        raise HTTPException(status_code=500, detail=f"Error deleting file record: {str(e)}")


@router.post("/files/bulk-delete")
@require_login
def bulk_delete_files(request: Request, file_ids: List[int], db: DbSession):
    """
    Delete multiple file records from the database.
    This only removes the database entries, not the actual files.
    """
    # Check if file deletion is allowed
    if not settings.allow_file_delete:
        raise HTTPException(status_code=403, detail="File deletion is disabled in the configuration")

    try:
        # Find all file records
        query = db.query(FileRecord).filter(FileRecord.id.in_(file_ids))
        query = apply_owner_filter(query, request)
        file_records = query.all()

        if not file_records:
            raise HTTPException(status_code=404, detail="No files found with the provided IDs")

        deleted_count = len(file_records)
        deleted_ids = [f.id for f in file_records]

        # Log the deletion
        logger.info(f"Bulk deleting {deleted_count} file records: IDs={deleted_ids}")

        # Delete all records
        for file_record in file_records:
            db.delete(file_record)

        db.commit()

        return {
            "status": "success",
            "message": f"Successfully deleted {deleted_count} file records",
            "deleted_ids": deleted_ids,
        }

    except HTTPException:
        raise
    except Exception as e:
        db.rollback()
        logger.exception(f"Error bulk deleting file records: {str(e)}")
        raise HTTPException(status_code=500, detail=f"Error bulk deleting file records: {str(e)}")


@router.post("/files/bulk-reprocess")
@require_login
def bulk_reprocess_files(request: Request, file_ids: List[int], db: DbSession):
    """
    Reprocess multiple files by queuing them for processing.
    """
    try:
        # Find all file records
        query = db.query(FileRecord).filter(FileRecord.id.in_(file_ids))
        query = apply_owner_filter(query, request)
        file_records = query.all()

        if not file_records:
            raise HTTPException(status_code=404, detail="No files found with the provided IDs")

        task_ids = []
        processed_files = []
        errors = []

        for file_record in file_records:
            try:
                # Check if local file exists
                if not file_record.local_filename or not os.path.exists(file_record.local_filename):
                    errors.append(
                        {
                            "file_id": file_record.id,
                            "filename": file_record.original_filename,
                            "error": "Local file not found",
                        }
                    )
                    continue

                # Queue the file for processing, passing file_id to skip duplicate check
                task = process_document.delay(file_record.local_filename, file_id=file_record.id)
                task_ids.append(task.id)
                processed_files.append(
                    {
                        "file_id": file_record.id,
                        "filename": file_record.original_filename,
                        "task_id": task.id,
                    }
                )

                logger.info(
                    f"Reprocessing file: ID={file_record.id}, "
                    f"Filename={file_record.original_filename}, TaskID={task.id}"
                )

            except Exception as e:
                logger.exception(f"Error reprocessing file {file_record.id}: {str(e)}")
                errors.append(
                    {
                        "file_id": file_record.id,
                        "filename": file_record.original_filename,
                        "error": str(e),
                    }
                )

        return {
            "status": "success" if processed_files else "error",
            "message": f"Successfully queued {len(processed_files)} files for reprocessing",
            "processed_files": processed_files,
            "errors": errors,
            "task_ids": task_ids,
        }

    except HTTPException:
        raise
    except Exception as e:
        logger.exception(f"Error bulk reprocessing files: {str(e)}")
        raise HTTPException(status_code=500, detail=f"Error bulk reprocessing files: {str(e)}")


@router.post("/files/bulk-reprocess-cloud-ocr")
@require_login
def bulk_reprocess_files_cloud_ocr(request: Request, file_ids: List[int], db: DbSession):
    """
    Reprocess multiple files with forced Cloud OCR (Azure Document Intelligence).

    Useful for re-running OCR on files with poor text quality or missing OCR text.
    """
    try:
        query = db.query(FileRecord).filter(FileRecord.id.in_(file_ids))
        query = apply_owner_filter(query, request)
        file_records = query.all()

        if not file_records:
            raise HTTPException(status_code=404, detail="No files found with the provided IDs")

        task_ids = []
        processed_files = []
        errors = []

        for file_record in file_records:
            try:
                source_file = None
                for path in [file_record.original_file_path, file_record.local_filename]:
                    if path and os.path.exists(path):
                        source_file = path
                        break

                if not source_file:
                    errors.append(
                        {
                            "file_id": file_record.id,
                            "filename": file_record.original_filename,
                            "error": "File not found on disk",
                        }
                    )
                    continue

                task = process_document.delay(
                    source_file,
                    original_filename=file_record.original_filename,
                    file_id=file_record.id,
                    force_cloud_ocr=True,
                )
                task_ids.append(task.id)
                processed_files.append(
                    {
                        "file_id": file_record.id,
                        "filename": file_record.original_filename,
                        "task_id": task.id,
                    }
                )
                logger.info(
                    f"Bulk Cloud OCR reprocessing: ID={file_record.id}, "
                    f"Filename={file_record.original_filename}, TaskID={task.id}"
                )
            except Exception as e:
                logger.exception(f"Error queuing Cloud OCR for file {file_record.id}: {str(e)}")
                errors.append(
                    {
                        "file_id": file_record.id,
                        "filename": file_record.original_filename,
                        "error": str(e),
                    }
                )

        return {
            "status": "success" if processed_files else "error",
            "message": f"Successfully queued {len(processed_files)} files for Cloud OCR reprocessing",
            "processed_files": processed_files,
            "errors": errors,
            "task_ids": task_ids,
        }

    except HTTPException:
        raise
    except Exception as e:
        logger.exception(f"Error bulk reprocessing files with Cloud OCR: {str(e)}")
        raise HTTPException(status_code=500, detail=f"Error bulk reprocessing files with Cloud OCR: {str(e)}")


@router.post("/files/bulk-download")
@require_login
def bulk_download_files(request: Request, file_ids: List[int], db: DbSession):
    """
    Download multiple files as a single ZIP archive.

    For each file, the processed version is preferred; falls back to the original.
    Files not found on disk are silently skipped.
    """
    try:
        query = db.query(FileRecord).filter(FileRecord.id.in_(file_ids))
        query = apply_owner_filter(query, request)
        file_records = query.all()

        if not file_records:
            raise HTTPException(status_code=404, detail="No files found with the provided IDs")

        zip_buffer = io.BytesIO()
        added = 0
        seen_names: dict[str, int] = {}

        with zipfile.ZipFile(zip_buffer, mode="w", compression=zipfile.ZIP_DEFLATED) as zf:
            workdir = settings.workdir
            processed_dir = os.path.join(workdir, "processed")

            for file_record in file_records:
                # Resolve file path: processed first, then original/local
                base_filename = os.path.splitext(file_record.original_filename or "file")[0]
                candidate_paths = [
                    file_record.processed_file_path,
                    file_record.original_file_path,
                    file_record.local_filename,
                    os.path.join(processed_dir, f"{file_record.filehash}.pdf"),
                    os.path.join(processed_dir, f"{base_filename}_processed.pdf"),
                ]

                file_path = None
                for path in candidate_paths:
                    if path and os.path.exists(path):
                        file_path = path
                        break

                if not file_path:
                    logger.warning(f"Skipping file {file_record.id}: no file found on disk")
                    continue

                # Build a unique archive name to avoid collisions
                archive_name = file_record.original_filename or os.path.basename(file_path)
                if archive_name in seen_names:
                    seen_names[archive_name] += 1
                    stem, ext = os.path.splitext(archive_name)
                    archive_name = f"{stem}_{seen_names[archive_name]}{ext}"
                else:
                    seen_names[archive_name] = 0

                zf.write(file_path, archive_name)
                added += 1

        if added == 0:
            raise HTTPException(status_code=404, detail="None of the selected files could be found on disk")

        zip_buffer.seek(0)
        timestamp = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
        zip_filename = f"docuelevate_bulk_{timestamp}.zip"

        logger.info(f"Bulk download: packed {added} file(s) into {zip_filename}")

        return StreamingResponse(
            zip_buffer,
            media_type="application/zip",
            headers={"Content-Disposition": f'attachment; filename="{zip_filename}"'},
        )

    except HTTPException:
        raise
    except Exception as e:
        logger.exception(f"Error creating bulk download ZIP: {str(e)}")
        raise HTTPException(status_code=500, detail=f"Error creating bulk download ZIP: {str(e)}")


@router.post("/files/{file_id}/reprocess")
@require_login
def reprocess_single_file(request: Request, file_id: int, db: DbSession):
    """
    Reprocess a single file by queuing it for processing again.

    Args:
        file_id: ID of the file to reprocess

    Returns:
        Task ID and status information
    """
    try:
        # Find the file record
        query = db.query(FileRecord).filter(FileRecord.id == file_id)
        query = apply_owner_filter(query, request)
        file_record = query.first()

        if not file_record:
            raise HTTPException(status_code=404, detail=f"File with ID {file_id} not found")

        # Check if local file exists
        if not file_record.local_filename or not os.path.exists(file_record.local_filename):
            raise HTTPException(
                status_code=400,
                detail="Local file not found on disk. Cannot reprocess.",
            )

        # Queue the file for processing, passing file_id to skip duplicate check
        task = process_document.delay(
            file_record.local_filename,
            original_filename=file_record.original_filename,
            file_id=file_record.id,
        )

        logger.info(
            f"Reprocessing file: ID={file_record.id}, Filename={file_record.original_filename}, TaskID={task.id}"
        )

        return {
            "status": "success",
            "message": "File queued for reprocessing",
            "file_id": file_record.id,
            "filename": file_record.original_filename,
            "task_id": task.id,
        }

    except HTTPException:
        raise
    except Exception as e:
        logger.exception(f"Error reprocessing file {file_id}: {str(e)}")
        raise HTTPException(status_code=500, detail=f"Error reprocessing file: {str(e)}")


@router.post("/files/{file_id}/reprocess-with-cloud-ocr")
@require_login
def reprocess_with_cloud_ocr(request: Request, file_id: int, db: DbSession):
    """
    Reprocess a single file with forced Cloud OCR processing.

    This endpoint forces Azure Document Intelligence OCR processing regardless
    of whether the PDF contains embedded text. Useful for documents with
    low-quality embedded text or when higher quality OCR is needed.

    Args:
        file_id: ID of the file to reprocess with Cloud OCR

    Returns:
        Task ID and status information
    """
    try:
        # Find the file record
        query = db.query(FileRecord).filter(FileRecord.id == file_id)
        query = apply_owner_filter(query, request)
        file_record = query.first()

        if not file_record:
            raise HTTPException(status_code=404, detail=f"File with ID {file_id} not found")

        # Prefer using the original_file_path if available, otherwise fall back to local_filename
        source_file = None
        if file_record.original_file_path and os.path.exists(file_record.original_file_path):
            source_file = file_record.original_file_path
            logger.info(f"Using original file for Cloud OCR reprocessing: {source_file}")
        elif file_record.local_filename and os.path.exists(file_record.local_filename):
            source_file = file_record.local_filename
            logger.info(f"Using local file for Cloud OCR reprocessing: {source_file}")
        else:
            raise HTTPException(
                status_code=400,
                detail="Neither original nor local file found on disk. Cannot reprocess.",
            )

        # Queue the file for processing with force_cloud_ocr=True
        task = process_document.delay(
            source_file,
            original_filename=file_record.original_filename,
            file_id=file_record.id,
            force_cloud_ocr=True,
        )

        logger.info(
            f"Reprocessing file with Cloud OCR: ID={file_record.id}, "
            f"Filename={file_record.original_filename}, TaskID={task.id}"
        )

        return {
            "status": "success",
            "message": "File queued for Cloud OCR reprocessing",
            "file_id": file_record.id,
            "filename": file_record.original_filename,
            "task_id": task.id,
            "force_cloud_ocr": True,
        }

    except HTTPException:
        raise
    except Exception as e:
        logger.exception(f"Error reprocessing file {file_id} with Cloud OCR: {str(e)}")
        raise HTTPException(status_code=500, detail=f"Error reprocessing file with Cloud OCR: {str(e)}")


def _extract_text_from_pdf(file_path: str) -> str:
    """
    Extract text from a PDF file using pypdf.

    Args:
        file_path: Path to the PDF file

    Returns:
        Extracted text from all pages
    """
    import pypdf  # Upgraded from PyPDF2 to fix CVE-2023-36464

    extracted_text = ""
    with open(file_path, "rb") as f:
        pdf_reader = pypdf.PdfReader(f)
        for page in pdf_reader.pages:
            extracted_text += page.extract_text() + "\n"
    return extracted_text


def _retry_pipeline_step(file_record: FileRecord, step_name: str, db: Session) -> dict:
    """
    Retry a specific pipeline processing step for a file.

    Supports restarting from intermediate pipeline steps:
    - process_document: Full reprocessing (skips duplicate check)
    - process_with_ocr: OCR processing (multi-provider)
    - process_with_azure_document_intelligence: OCR processing (legacy alias for process_with_ocr)
    - extract_metadata_with_gpt: Metadata extraction
    - embed_metadata_into_pdf: Metadata embedding

    Args:
        file_record: The FileRecord to reprocess
        step_name: Name of the pipeline step to retry
        db: Database session

    Returns:
        Dict with task ID and status information
    """
    file_id = file_record.id

    if step_name == "process_document":
        # Full reprocessing with duplicate check bypass
        logger.info(f"Retrying process_document for file {file_id}: local_filename={file_record.local_filename!r}")
        if not file_record.local_filename:
            logger.error(f"process_document retry failed for file {file_id}: local_filename is None")
            raise HTTPException(status_code=400, detail="Local file path is None. Cannot retry.")

        exists = os.path.exists(file_record.local_filename)
        logger.info(f"Checking local_filename: {file_record.local_filename!r}, exists={exists}")
        if not exists:
            error_message = f"Local file not found on disk. Cannot retry. Path checked: local_filename={file_record.local_filename!r} (exists=False)"
            logger.error(f"process_document retry failed for file {file_id}: {error_message}")
            raise HTTPException(status_code=400, detail=error_message)

        logger.info(f"Found file for process_document retry at: {file_record.local_filename!r}")
        task = process_document.delay(
            file_record.local_filename,
            original_filename=file_record.original_filename,
            file_id=file_id,
        )
    elif step_name in ("process_with_ocr", "process_with_azure_document_intelligence"):
        from app.tasks.process_with_ocr import process_with_ocr

        # OCR needs the file in workdir/tmp
        logger.info(f"Retrying process_with_ocr for file {file_id}: local_filename={file_record.local_filename!r}")
        if not file_record.local_filename:
            logger.error(f"OCR retry failed for file {file_id}: local_filename is None")
            raise HTTPException(status_code=400, detail="Local file path is None. Cannot retry OCR.")

        exists = os.path.exists(file_record.local_filename)
        logger.info(f"Checking local_filename: {file_record.local_filename!r}, exists={exists}")
        if not exists:
            error_message = f"Local file not found on disk. Cannot retry OCR. Path checked: local_filename={file_record.local_filename!r} (exists=False)"
            logger.error(f"OCR retry failed for file {file_id}: {error_message}")
            raise HTTPException(status_code=400, detail=error_message)

        logger.info(f"Found file for OCR retry at: {file_record.local_filename!r}")
        filename = os.path.basename(file_record.local_filename)
        task = process_with_ocr.delay(filename, file_id)
    elif step_name == "extract_metadata_with_gpt":
        from app.tasks.extract_metadata_with_gpt import extract_metadata_with_gpt

        logger.info(
            f"Retrying extract_metadata_with_gpt for file {file_id}: local_filename={file_record.local_filename!r}"
        )
        if not file_record.local_filename:
            logger.error(f"Metadata extraction retry failed for file {file_id}: local_filename is None")
            raise HTTPException(
                status_code=400,
                detail="Local file path is None. Cannot retry metadata extraction.",
            )

        exists = os.path.exists(file_record.local_filename)
        logger.info(f"Checking local_filename: {file_record.local_filename!r}, exists={exists}")
        if not exists:
            error_message = f"Local file not found on disk. Cannot retry metadata extraction. Path checked: local_filename={file_record.local_filename!r} (exists=False)"
            logger.error(f"Metadata extraction retry failed for file {file_id}: {error_message}")
            raise HTTPException(status_code=400, detail=error_message)

        logger.info(f"Found file for metadata extraction retry at: {file_record.local_filename!r}")
        extracted_text = _extract_text_from_pdf(file_record.local_filename)
        filename = os.path.basename(file_record.local_filename)
        task = extract_metadata_with_gpt.delay(filename, extracted_text, file_id)
    elif step_name == "embed_metadata_into_pdf":
        from app.tasks.extract_metadata_with_gpt import (
            extract_metadata_with_gpt as extract_metadata_task,
        )

        # Retrying embed requires re-running metadata extraction first, because
        # embed_metadata_into_pdf needs the actual metadata dict (not empty).
        # Re-trigger extract_metadata_with_gpt which will chain into embed_metadata_into_pdf.

        # Log all database path values at the start
        logger.info(
            f"Retrying embed_metadata_into_pdf for file {file_id}: "
            f"local_filename={file_record.local_filename!r}, "
            f"processed_file_path={file_record.processed_file_path!r}, "
            f"original_file_path={file_record.original_file_path!r}"
        )

        # Check for file in multiple locations:
        # 1. Original location in tmp (file_record.local_filename)
        # 2. Processed location (file_record.processed_file_path)
        # 3. Original immutable copy (file_record.original_file_path)
        # 4. Fallback to workdir/tmp/<basename>
        file_path = None
        checked_paths = []

        # Check 1: local_filename (original tmp location)
        if file_record.local_filename:
            exists = os.path.exists(file_record.local_filename)
            checked_paths.append(f"local_filename={file_record.local_filename!r} (exists={exists})")
            logger.info(f"Checking local_filename: {file_record.local_filename!r}, exists={exists}")
            if exists:
                file_path = file_record.local_filename

        # Check 2: processed_file_path
        if not file_path and file_record.processed_file_path:
            exists = os.path.exists(file_record.processed_file_path)
            checked_paths.append(f"processed_file_path={file_record.processed_file_path!r} (exists={exists})")
            logger.info(f"Checking processed_file_path: {file_record.processed_file_path!r}, exists={exists}")
            if exists:
                file_path = file_record.processed_file_path

        # Check 3: original_file_path (immutable copy in workdir/original/)
        if not file_path and file_record.original_file_path:
            exists = os.path.exists(file_record.original_file_path)
            checked_paths.append(f"original_file_path={file_record.original_file_path!r} (exists={exists})")
            logger.info(f"Checking original_file_path: {file_record.original_file_path!r}, exists={exists}")
            if exists:
                file_path = file_record.original_file_path

        # Check 4: Fallback to workdir/tmp/<basename>
        if not file_path and file_record.local_filename:
            workdir = settings.workdir
            tmp_dir = os.path.join(workdir, "tmp")
            fallback_path = os.path.join(tmp_dir, os.path.basename(file_record.local_filename))
            exists = os.path.exists(fallback_path)
            checked_paths.append(f"workdir_tmp_fallback={fallback_path!r} (exists={exists})")
            logger.info(f"Checking workdir/tmp fallback: {fallback_path!r}, exists={exists}")
            if exists:
                file_path = fallback_path

        if not file_path:
            paths_detail = "; ".join(checked_paths)
            error_message = f"File not found on disk. Cannot retry metadata embedding. Paths checked: {paths_detail}"
            logger.error(f"embed_metadata_into_pdf retry failed for file {file_id}: {error_message}")
            raise HTTPException(status_code=400, detail=error_message)

        logger.info(f"Found file for embed_metadata_into_pdf retry at: {file_path!r}")
        extracted_text = _extract_text_from_pdf(file_path)
        # Pass the full path to the task so it can locate the file
        task = extract_metadata_task.delay(file_path, extracted_text, file_id)
    else:
        raise HTTPException(status_code=400, detail=f"Unsupported pipeline step: {step_name}")

    logger.info(f"Retrying pipeline step: FileID={file_record.id}, Step={step_name}, TaskID={task.id}")

    return {
        "status": "success",
        "message": f"Pipeline step {step_name} queued for retry",
        "file_id": file_record.id,
        "subtask_name": step_name,
        "task_id": task.id,
    }


@router.post("/files/{file_id}/retry-subtask")
@require_login
def retry_subtask(
    request: Request,
    file_id: int,
    db: DbSession,
    subtask_name: str = Query(
        ...,
        description="Name of the subtask to retry (e.g., 'upload_to_dropbox', 'extract_metadata_with_gpt')",
    ),
):
    """
    Retry a specific failed subtask for a file.

    Supports both upload tasks (e.g., upload_to_dropbox) and pipeline processing
    steps (e.g., process_with_ocr, extract_metadata_with_gpt,
    embed_metadata_into_pdf).

    Args:
        file_id: ID of the file
        subtask_name: Name of the task to retry

    Returns:
        Task ID and status information
    """
    try:
        # Find the file record
        query = db.query(FileRecord).filter(FileRecord.id == file_id)
        query = apply_owner_filter(query, request)
        file_record = query.first()

        if not file_record:
            raise HTTPException(status_code=404, detail=f"File with ID {file_id} not found")

        # Pipeline processing steps that can be retried from the failed step
        pipeline_step_names = {
            "process_document",
            "process_with_ocr",
            "process_with_azure_document_intelligence",
            "extract_metadata_with_gpt",
            "embed_metadata_into_pdf",
        }

        if subtask_name in pipeline_step_names:
            return _retry_pipeline_step(file_record, subtask_name, db)

        # Map subtask names to their corresponding Celery tasks
        from app.tasks.upload_to_dropbox import upload_to_dropbox
        from app.tasks.upload_to_email import upload_to_email
        from app.tasks.upload_to_ftp import upload_to_ftp
        from app.tasks.upload_to_google_drive import upload_to_google_drive
        from app.tasks.upload_to_nextcloud import upload_to_nextcloud
        from app.tasks.upload_to_onedrive import upload_to_onedrive
        from app.tasks.upload_to_paperless import upload_to_paperless
        from app.tasks.upload_to_s3 import upload_to_s3
        from app.tasks.upload_to_sftp import upload_to_sftp
        from app.tasks.upload_to_webdav import upload_to_webdav

        task_map = {
            "upload_to_dropbox": upload_to_dropbox,
            "upload_to_nextcloud": upload_to_nextcloud,
            "upload_to_paperless": upload_to_paperless,
            "upload_to_google_drive": upload_to_google_drive,
            "upload_to_onedrive": upload_to_onedrive,
            "upload_to_s3": upload_to_s3,
            "upload_to_webdav": upload_to_webdav,
            "upload_to_ftp": upload_to_ftp,
            "upload_to_sftp": upload_to_sftp,
            "upload_to_email": upload_to_email,
        }

        if subtask_name not in task_map:
            all_valid = sorted(list(task_map.keys()) + sorted(pipeline_step_names))
            raise HTTPException(
                status_code=400,
                detail=f"Invalid subtask name: {subtask_name}. Must be one of: {', '.join(all_valid)}",
            )

        # Check for processed file (upload tasks work with processed files)
        logger.info(
            f"Retrying upload task {subtask_name} for file {file_id}: "
            f"original_filename={file_record.original_filename!r}, "
            f"filehash={file_record.filehash!r}, "
            f"processed_file_path={file_record.processed_file_path!r}"
        )

        # Try to find the processed file using multiple strategies:
        # 1. DB-stored processed_file_path (most reliable - set during finalization)
        # 2. Hash-based path in processed dir
        # 3. Original filename with _processed suffix
        # 4. Original filename in processed dir
        file_path = None
        checked_paths = []

        # Check 1: processed_file_path from DB
        if file_record.processed_file_path:
            exists = os.path.exists(file_record.processed_file_path)
            checked_paths.append(f"processed_file_path={file_record.processed_file_path!r} (exists={exists})")
            logger.info(f"Checking processed_file_path: {file_record.processed_file_path!r}, exists={exists}")
            if exists:
                file_path = file_record.processed_file_path

        # Check 2-4: Legacy path patterns in processed directory
        if not file_path:
            workdir = settings.workdir
            processed_dir = os.path.join(workdir, "processed")
            base_filename = os.path.splitext(file_record.original_filename)[0]
            legacy_paths = [
                os.path.join(processed_dir, f"{file_record.filehash}.pdf"),
                os.path.join(processed_dir, f"{base_filename}_processed.pdf"),
                os.path.join(processed_dir, file_record.original_filename),
            ]

            for path in legacy_paths:
                exists = os.path.exists(path)
                checked_paths.append(f"{path!r} (exists={exists})")
                logger.info(f"Checking processed file path: {path!r}, exists={exists}")
                if exists:
                    file_path = path
                    break

        if not file_path:
            paths_detail = "; ".join(checked_paths)
            error_message = f"Processed file not found. Cannot retry upload. Paths checked: {paths_detail}"
            logger.error(f"Upload retry failed for file {file_id}: {error_message}")
            raise HTTPException(status_code=400, detail=error_message)

        logger.info(f"Found processed file for upload retry at: {file_path!r}")
        # Queue the specific upload task
        upload_task = task_map[subtask_name]
        task = upload_task.delay(file_path, file_id)

        logger.info(f"Retrying upload subtask: FileID={file_record.id}, Subtask={subtask_name}, TaskID={task.id}")

        return {
            "status": "success",
            "message": f"Upload task {subtask_name} queued for retry",
            "file_id": file_record.id,
            "subtask_name": subtask_name,
            "task_id": task.id,
        }

    except HTTPException:
        raise
    except Exception as e:
        logger.exception(f"Error retrying subtask {subtask_name} for file {file_id}: {str(e)}")
        raise HTTPException(status_code=500, detail=f"Error retrying subtask: {str(e)}")


@router.get("/files/{file_id}/preview")
@require_login
def get_file_preview(
    request: Request,
    file_id: int,
    db: DbSession,
    version: str = Query("original", description="original or processed"),
):
    """
    Get file content for preview (original or processed version).

    Args:
        file_id: ID of the file
        version: "original" for tmp file, "processed" for processed file

    Returns:
        File content for preview
    """
    from fastapi.responses import FileResponse

    try:
        # Find the file record
        query = db.query(FileRecord).filter(FileRecord.id == file_id)
        query = apply_owner_filter(query, request)
        file_record = query.first()

        if not file_record:
            raise HTTPException(status_code=404, detail=f"File with ID {file_id} not found")

        if version == "original":
            # Use the stored original_file_path first (persisted copy), then fall back to local_filename (tmp)
            file_path = None
            for path in [file_record.original_file_path, file_record.local_filename]:
                if path and os.path.exists(path):
                    file_path = path
                    break

            if not file_path:
                raise HTTPException(status_code=404, detail="Original file not found on disk")

        elif version == "processed":
            # Use the stored processed_file_path first, then fall back to guessing
            workdir = settings.workdir
            processed_dir = os.path.join(workdir, "processed")

            # Try to find the processed file (stored path first, then hash or UUID-based naming)
            base_filename = os.path.splitext(file_record.original_filename)[0]
            potential_paths = [
                file_record.processed_file_path,
                os.path.join(processed_dir, f"{file_record.filehash}.pdf"),
                os.path.join(processed_dir, f"{base_filename}_processed.pdf"),
                os.path.join(processed_dir, file_record.original_filename),
            ]

            file_path = None
            for path in potential_paths:
                if path and os.path.exists(path):
                    file_path = path
                    break

            if not file_path:
                raise HTTPException(status_code=404, detail="Processed file not found")
        else:
            raise HTTPException(
                status_code=400,
                detail="Invalid version parameter. Use 'original' or 'processed'",
            )

        # Return the file
        return FileResponse(
            path=file_path,
            media_type=file_record.mime_type or "application/pdf",
            headers={"Content-Disposition": f'inline; filename="{file_record.original_filename}"'},
        )

    except HTTPException:
        raise
    except Exception as e:
        logger.exception(f"Error retrieving file preview: {str(e)}")
        raise HTTPException(status_code=500, detail=f"Error retrieving file preview: {str(e)}")


@router.get("/files/{file_id}/download")
@require_login
def download_file(
    request: Request,
    file_id: int,
    db: DbSession,
    version: str = Query("processed", description="'processed' (default) or 'original'"),
):
    """
    Download file (processed or original version) as attachment.

    Args:
        file_id: ID of the file
        version: "processed" (default) for the post-processing file, "original" for the raw upload

    Returns:
        File content as attachment download
    """
    from fastapi.responses import FileResponse

    try:
        # Find the file record
        query = db.query(FileRecord).filter(FileRecord.id == file_id)
        query = apply_owner_filter(query, request)
        file_record = query.first()

        if not file_record:
            raise HTTPException(status_code=404, detail=f"File with ID {file_id} not found")

        if version == "original":
            # Use the stored original_file_path first (persisted copy), then fall back to local_filename (tmp)
            file_path = None
            for path in [file_record.original_file_path, file_record.local_filename]:
                if path and os.path.exists(path):
                    file_path = path
                    break

            if not file_path:
                raise HTTPException(status_code=404, detail="Original file not found on disk")

        elif version == "processed":
            # Use the stored processed_file_path first, then fall back to guessing
            workdir = settings.workdir
            processed_dir = os.path.join(workdir, "processed")

            # Try to find the processed file (stored path first, then hash or UUID-based naming)
            base_filename = os.path.splitext(file_record.original_filename)[0]
            potential_paths = [
                file_record.processed_file_path,
                os.path.join(processed_dir, f"{file_record.filehash}.pdf"),
                os.path.join(processed_dir, f"{base_filename}_processed.pdf"),
                os.path.join(processed_dir, file_record.original_filename),
            ]

            file_path = None
            for path in potential_paths:
                if path and os.path.exists(path):
                    file_path = path
                    break

            if not file_path:
                raise HTTPException(status_code=404, detail="Processed file not found")
        else:
            raise HTTPException(
                status_code=400,
                detail="Invalid version parameter. Use 'original' or 'processed'",
            )

        # Return the file with attachment disposition to trigger download
        return FileResponse(
            path=file_path,
            media_type=file_record.mime_type or "application/pdf",
            headers={"Content-Disposition": f'attachment; filename="{file_record.original_filename}"'},
        )

    except HTTPException:
        raise
    except Exception as e:
        logger.exception(f"Error downloading file: {str(e)}")
        raise HTTPException(status_code=500, detail=f"Error downloading file: {str(e)}")


async def _save_upload_file_chunks(file: UploadFile, target_path: str, max_size: int) -> int:
    """Save an uploaded file in chunks and enforce the maximum size limit."""
    try:
        written_size = 0
        with open(target_path, "wb") as f:
            chunk_size = 65536  # 64 KB chunks
            while True:
                chunk = await file.read(chunk_size)
                if not chunk:
                    break
                written_size += len(chunk)
                if written_size > max_size:
                    # Exceeded limit mid-stream; clean up and reject
                    f.close()
                    os.remove(target_path)
                    raise HTTPException(
                        status_code=413,
                        detail=f"File too large: exceeded {max_size} bytes during upload. "
                        f"See SECURITY_AUDIT.md for configuration details.",
                    )
                f.write(chunk)
        return written_size
    except HTTPException:
        raise
    except Exception as e:
        if os.path.exists(target_path):
            os.remove(target_path)
        raise HTTPException(status_code=500, detail=f"Failed to save file: {e}")


def _check_for_exact_duplicate(db: DbSession, target_path: str, safe_filename: str) -> dict | None:
    """Check for an exact duplicate of the uploaded file.

    Returns a dict with duplicate info when the file's SHA-256 hash matches an
    already-processed document, or ``None`` when no duplicate is found (or
    deduplication is disabled).
    """
    if not settings.enable_deduplication:
        return None

    try:
        filehash = hash_file(target_path)
        existing = (
            db.query(FileRecord)
            .filter(FileRecord.filehash == filehash, FileRecord.is_duplicate.is_(False))
            .order_by(FileRecord.id.asc())
            .first()
        )
        if existing:
            logger.info(f"Exact duplicate detected on upload: '{safe_filename}' matches file ID {existing.id}")
            return {
                "duplicate_type": "exact",
                "original_file_id": existing.id,
                "original_filename": existing.original_filename,
                "message": (
                    "This file is an exact duplicate of an already-processed document. "
                    "It has not been queued for processing again."
                ),
            }
    except Exception as e:
        logger.warning(f"Duplicate check failed for uploaded file '{safe_filename}': {e}")

    return None


@router.post("/ui-upload")
@require_login
async def ui_upload(
    request: Request,
    db: DbSession,
    file: UploadFile = File(...),
    _rate_ok: None = Depends(require_upload_rate_limit),
):
    """Endpoint to accept a user-uploaded file and enqueue it for processing."""
    workdir = settings.workdir

    # Early size check: reject before reading the body if Content-Length is known
    max_size = settings.max_upload_size
    content_length_header = request.headers.get("content-length")
    if content_length_header is not None:
        try:
            declared_size = int(content_length_header)
            if declared_size > max_size:
                raise HTTPException(
                    status_code=413,
                    detail=f"File too large: declared size {declared_size} bytes exceeds maximum "
                    f"{max_size} bytes. See SECURITY_AUDIT.md for configuration details.",
                )
        except ValueError:
            pass  # Malformed header; proceed and check actual size after reading

    # Extract just the filename without any path components to prevent path traversal
    # First, use basename to remove any directory components
    base_filename = os.path.basename(file.filename)
    # Then sanitize the filename to remove special characters and ensure filesystem compatibility
    safe_filename = sanitize_filename(base_filename)

    # Generate a unique filename with UUID to prevent overwriting and filename conflicts
    unique_id = str(uuid.uuid4())
    # Keep the original extension if present
    if "." in safe_filename:
        file_extension = safe_filename.rsplit(".", 1)[1]
        target_filename = f"{unique_id}.{file_extension}"
    else:
        target_filename = unique_id

    # Store both the safe original name and the unique name
    target_path = os.path.join(workdir, target_filename)

    # Determine the owner_id for multi-user document isolation
    upload_owner_id = get_current_owner_id(request) if settings.multi_user_enabled else None

    # Enforce subscription tier upload quotas (multi-user mode only) BEFORE writing the file
    # so that users who have exceeded their quota do not waste bandwidth or disk I/O.
    if settings.multi_user_enabled and upload_owner_id:
        from app.utils.subscription import QuotaExceeded, check_upload_allowed, get_user_tier_id

        tier_id = get_user_tier_id(db, upload_owner_id)
        try:
            check_upload_allowed(db, upload_owner_id, tier_id)
        except QuotaExceeded as qe:
            raise HTTPException(
                status_code=status.HTTP_402_PAYMENT_REQUIRED,
                detail=str(qe),
            )

    # Read file in chunks to avoid loading the entire body into memory at once,
    # enforcing the size limit during the read so memory usage stays bounded.
    try:
        written_size = 0
        async with aiofiles.open(target_path, "wb") as f:
            chunk_size = 65536  # 64 KB chunks
            while True:
                chunk = await file.read(chunk_size)
                if not chunk:
                    break
                written_size += len(chunk)
                if written_size > max_size:
                    # Exceeded limit mid-stream; clean up and reject
                    await f.close()
                    os.remove(target_path)
                    raise HTTPException(
                        status_code=413,
                        detail=f"File too large: exceeded {max_size} bytes during upload. "
                        f"See SECURITY_AUDIT.md for configuration details.",
                    )
                await f.write(chunk)
    except HTTPException:
        raise
    except Exception as e:
        if os.path.exists(target_path):
            os.remove(target_path)
        raise HTTPException(status_code=500, detail=f"Failed to save file: {e}")

    # Log the mapping between original and safe filename
    logger.info(f"Saved uploaded file '{safe_filename}' as '{target_filename}'")
    file_size = written_size

    # ── Early duplicate rejection ──────────────────────────────────────────
    # Check for exact duplicates (same SHA-256 hash) BEFORE enqueuing a
    # processing task.  When deduplication is enabled and the file already
    # exists, we skip processing entirely, clean up the temp file, and
    # return the existing file's information to the caller.
    exact_duplicate = _check_for_exact_duplicate(db, target_path, safe_filename)
    if exact_duplicate:
        # Remove the just-saved temp file — it's a duplicate.
        try:
            os.remove(target_path)
        except OSError:
            pass
        return {
            "status": "duplicate",
            "original_filename": safe_filename,
            "stored_filename": target_filename,
            "duplicate_of": exact_duplicate,
        }

    # Determine if the file is a PDF or needs conversion
    mime_type, _ = mimetypes.guess_type(target_path)
    file_ext = os.path.splitext(target_path)[1].lower()

    # Check if it's a PDF by extension or MIME type
    is_pdf = file_ext == ".pdf" or mime_type == "application/pdf"

    # Check if file splitting is needed (only for PDFs)
    from app.utils.file_splitting import should_split_file

    should_split = is_pdf and should_split_file(target_path, settings.max_single_file_size)

    if should_split:
        # File needs to be split before processing
        from app.utils.file_splitting import split_pdf_by_size

        try:
            logger.info(
                f"File {target_path} ({file_size} bytes) exceeds max_single_file_size "
                f"({settings.max_single_file_size} bytes). Splitting..."
            )
            split_files = split_pdf_by_size(target_path, settings.max_single_file_size)
            logger.info(f"Split {target_path} into {len(split_files)} parts")

            # Queue each split file for processing
            task_ids = []
            for split_file in split_files:
                split_filename = os.path.basename(split_file)
                task = process_document.delay(split_file, original_filename=split_filename, owner_id=upload_owner_id)
                task_ids.append(task.id)
                logger.info(f"Enqueued split PDF part for processing: {split_file}")

            # Remove the original file after successful splitting
            os.remove(target_path)

            return {
                "task_ids": task_ids,
                "status": "queued",
                "original_filename": safe_filename,
                "stored_filename": target_filename,
                "split_into_parts": len(split_files),
                "message": f"File split into {len(split_files)} parts for processing",
            }
        except Exception as e:
            logger.exception(f"Failed to split file {target_path}: {str(e)}")
            # Fall back to processing the whole file
            logger.warning(f"Falling back to processing whole file due to split error: {str(e)}")
            should_split = False

    if is_pdf and not should_split:
        # If it's a PDF, process directly
        task = process_document.delay(target_path, original_filename=safe_filename, owner_id=upload_owner_id)
        logger.info(f"Enqueued PDF for processing: {target_path}")
    elif mime_type in IMAGE_MIME_TYPES or file_ext in {
        ".jpg",
        ".jpeg",
        ".png",
        ".gif",
        ".bmp",
        ".tiff",
        ".tif",
        ".webp",
        ".svg",
        ".heic",
        ".heif",
    }:
        # If it's an image, convert to PDF first
        task = convert_to_pdf.delay(target_path, original_filename=safe_filename, owner_id=upload_owner_id)
        logger.info(f"Enqueued image for PDF conversion: {target_path}")
    elif mime_type in ALLOWED_MIME_TYPES or file_ext in ALLOWED_EXTENSIONS:
        # Office document, HTML, Markdown, or other Gotenberg-supported format
        task = convert_to_pdf.delay(target_path, original_filename=safe_filename, owner_id=upload_owner_id)
        logger.info(f"Enqueued document for PDF conversion: {target_path}")
    else:
        # For any other file type, attempt conversion but log a warning
        logger.warning(f"Unsupported MIME type {mime_type} for {target_path}, attempting conversion")
        task = convert_to_pdf.delay(target_path, original_filename=safe_filename, owner_id=upload_owner_id)

    return {
        "task_id": task.id,
        "status": "queued",
        "original_filename": safe_filename,
        "stored_filename": target_filename,
    }


# ---------------------------------------------------------------------------
# Document ownership / claim endpoints
# ---------------------------------------------------------------------------


@router.post("/files/{file_id}/claim")
@require_login
def claim_file(request: Request, file_id: int, db: DbSession):
    """
    Claim an unowned document for the current user.

    Only documents with ``owner_id IS NULL`` can be claimed.  The requesting
    user's identifier is written into ``owner_id``.  In single-user mode
    the endpoint is a no-op (returns the file unchanged).
    """
    if not settings.multi_user_enabled:
        raise HTTPException(status_code=400, detail="Multi-user mode is not enabled")

    owner_id = get_current_owner_id(request)
    if owner_id is None:
        raise HTTPException(status_code=401, detail="Authentication required to claim a document")

    file_record = db.query(FileRecord).filter(FileRecord.id == file_id).first()
    if not file_record:
        raise HTTPException(status_code=404, detail=f"File record with ID {file_id} not found")

    if file_record.owner_id is not None:
        if file_record.owner_id == owner_id:
            return {"status": "already_owned", "message": "You already own this document", "file_id": file_id}
        raise HTTPException(status_code=403, detail="This document is already owned by another user")

    file_record.owner_id = owner_id
    try:
        db.commit()
    except Exception as e:
        db.rollback()
        logger.exception(f"Error claiming file {file_id}: {e}")
        raise HTTPException(status_code=500, detail="Failed to claim document")

    logger.info("File %d claimed by user", file_id)
    return {"status": "success", "message": "Document claimed successfully", "file_id": file_id, "owner_id": owner_id}


@router.post("/files/bulk-claim")
@require_login
def bulk_claim_files(request: Request, file_ids: list[int], db: DbSession):
    """
    Claim multiple unowned documents for the current user.

    Only documents with ``owner_id IS NULL`` will be claimed.  Documents
    already owned (by anyone) are skipped and reported in ``skipped``.
    """
    if not settings.multi_user_enabled:
        raise HTTPException(status_code=400, detail="Multi-user mode is not enabled")

    owner_id = get_current_owner_id(request)
    if owner_id is None:
        raise HTTPException(status_code=401, detail="Authentication required to claim documents")

    file_records = db.query(FileRecord).filter(FileRecord.id.in_(file_ids)).all()
    if not file_records:
        raise HTTPException(status_code=404, detail="No files found with the provided IDs")

    claimed = []
    skipped = []
    for rec in file_records:
        if rec.owner_id is None:
            rec.owner_id = owner_id
            claimed.append(rec.id)
        else:
            skipped.append({"file_id": rec.id, "reason": "already owned"})

    try:
        db.commit()
    except Exception as e:
        db.rollback()
        logger.exception(f"Error during bulk claim: {e}")
        raise HTTPException(status_code=500, detail="Failed to claim documents")

    logger.info("Bulk claim: claimed=%s, skipped=%s", claimed, [s["file_id"] for s in skipped])
    return {
        "status": "success",
        "claimed_count": len(claimed),
        "claimed_ids": claimed,
        "skipped": skipped,
        "owner_id": owner_id,
    }


@router.post("/files/assign-owner")
@require_login
def assign_owner(request: Request, db: DbSession, owner_id: str = Query(...), file_ids: list[int] | None = None):
    """
    Admin-only: assign an owner to documents.

    If ``file_ids`` is provided, only those files are updated.  If omitted,
    **all** currently unowned documents (``owner_id IS NULL``) are assigned
    to the given ``owner_id``.
    """
    if not settings.multi_user_enabled:
        raise HTTPException(status_code=400, detail="Multi-user mode is not enabled")

    user = request.session.get("user")
    if not isinstance(user, dict) or not user.get("is_admin"):
        raise HTTPException(status_code=403, detail="Only admins can assign document owners")

    if not owner_id or not owner_id.strip():
        raise HTTPException(status_code=422, detail="owner_id must be a non-empty string")
    owner_id = owner_id.strip()

    if file_ids is not None:
        # Assign to specific files
        updated = (
            db.query(FileRecord)
            .filter(FileRecord.id.in_(file_ids))
            .update({FileRecord.owner_id: owner_id}, synchronize_session="fetch")
        )
    else:
        # Assign to all currently unowned documents
        updated = (
            db.query(FileRecord)
            .filter(FileRecord.owner_id.is_(None))
            .update({FileRecord.owner_id: owner_id}, synchronize_session="fetch")
        )

    try:
        db.commit()
    except Exception as e:
        db.rollback()
        logger.exception(f"Error assigning owner: {e}")
        raise HTTPException(status_code=500, detail="Failed to assign owner")

    logger.info("Admin assigned owner to %d file(s)", updated)
    return {
        "status": "success",
        "message": f"Assigned owner to {updated} document(s)",
        "updated_count": updated,
        "owner_id": owner_id,
    }


# ---------------------------------------------------------------------------
# Pipeline assignment
# ---------------------------------------------------------------------------


@router.post("/files/{file_id}/assign-pipeline")
@require_login
def assign_pipeline_to_file(
    request: Request,
    file_id: int,
    db: DbSession,
    pipeline_id: int | None = None,
):
    """Assign (or remove) a processing pipeline from a file.

    Path Parameters:
        file_id: The file to update.

    Query / Body Parameters:
        pipeline_id: The pipeline to assign.  Pass ``null`` or omit to clear the
            assignment (the system default will be used for future processing).

    Returns:
        A summary dict with the file_id and updated pipeline_id.

    Raises:
        HTTPException 404: If the file or pipeline does not exist / is not
            accessible to the current user.
    """
    from app.auth import get_current_user, get_current_user_id
    from app.models import Pipeline

    user = get_current_user(request)
    user_id: str = get_current_user_id(request)

    is_admin_user = bool(user and user.get("is_admin"))

    file_record = db.query(FileRecord).filter(FileRecord.id == file_id).first()
    if not file_record:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found")

    # Non-admins may only update files they own (or unowned files in single-user mode)
    owner_id = get_current_owner_id(request)
    if not is_admin_user and file_record.owner_id is not None and file_record.owner_id != owner_id:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found")

    if pipeline_id is not None:
        pipeline = db.query(Pipeline).filter(Pipeline.id == pipeline_id).first()
        if not pipeline:
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Pipeline not found")
        # Check access: users can only assign their own pipelines or system pipelines (owner_id=None)
        if not is_admin_user and pipeline.owner_id is not None and pipeline.owner_id != user_id:
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Pipeline not found")

    file_record.pipeline_id = pipeline_id

    try:
        db.commit()
        db.refresh(file_record)
    except Exception as exc:
        db.rollback()
        logger.exception(f"Failed to assign pipeline to file id={file_id}: {exc}")
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to assign pipeline")

    logger.info(f"Pipeline {pipeline_id!r} assigned to file id={file_id}")
    return {"file_id": file_id, "pipeline_id": file_record.pipeline_id}