feat(multi-user): add multi-user feature flag, owner_id model field, and user-scoped queries

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>

.env.demo

@@ -129,6 +129,13 @@ ADMIN_USERNAME=admin
 ADMIN_PASSWORD=your_secure_password
 ADMIN_GROUP_NAME=admin
 
+# **Multi-User Mode**
+# When enabled, each user has their own document space with isolated uploads,
+# search, and file management. Requires AUTH_ENABLED=true.
+MULTI_USER_ENABLED=false
+# Default upload limit per user per day (0 = unlimited)
+DEFAULT_DAILY_UPLOAD_LIMIT=0
+
 # **OpenID Connect/Authentik Settings**
 AUTHENTIK_CLIENT_ID=<yourAuthentikAppClientID>
 AUTHENTIK_CLIENT_SECRET=<yourAuthentikClientSecret>

app/api/files.py

@@ -99,8 +99,11 @@ def list_files_api(
     validate_sort_order(sort_order)
     search = validate_search_query(search)
 
-    # Start with base query
+    # Start with base query, scoped to the current user in multi-user mode
+    from app.utils.user_scope import apply_owner_filter
+
     query = db.query(FileRecord)
+    query = apply_owner_filter(query, request)
 
     # Apply search filter
     if search:
@@ -241,8 +244,12 @@ 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
-    file_record = db.query(FileRecord).filter(FileRecord.id == file_id).first()
+    # Find the file record, scoped to the current user in multi-user mode
+    from app.utils.user_scope import apply_owner_filter
+
+    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")
@@ -300,8 +307,12 @@ def delete_file_record(request: Request, file_id: int, db: DbSession):
         raise HTTPException(status_code=403, detail="File deletion is disabled in the configuration")
 
     try:
-        # Find the file record
-        file_record = db.query(FileRecord).filter(FileRecord.id == file_id).first()
+        # Find the file record, scoped to the current user in multi-user mode
+        from app.utils.user_scope import apply_owner_filter
+
+        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")
@@ -1286,6 +1297,11 @@ async def ui_upload(request: Request, db: DbSession, file: UploadFile = File(...
     mime_type, _ = mimetypes.guess_type(target_path)
     file_ext = os.path.splitext(target_path)[1].lower()
 
+    # Determine the owner_id for multi-user document isolation
+    from app.utils.user_scope import get_current_owner_id
+
+    upload_owner_id = get_current_owner_id(request) if settings.multi_user_enabled else None
+
     # Check if it's a PDF by extension or MIME type
     is_pdf = file_ext == ".pdf" or mime_type == "application/pdf"
 
@@ -1310,7 +1326,7 @@ async def ui_upload(request: Request, db: DbSession, file: UploadFile = File(...
             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)
+                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}")
 
@@ -1333,7 +1349,7 @@ async def ui_upload(request: Request, db: DbSession, file: UploadFile = File(...
 
     if is_pdf and not should_split:
         # If it's a PDF, process directly
-        task = process_document.delay(target_path, original_filename=safe_filename)
+        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",
@@ -1347,16 +1363,16 @@ async def ui_upload(request: Request, db: DbSession, file: UploadFile = File(...
         ".svg",
     }:
         # If it's an image, convert to PDF first
-        task = convert_to_pdf.delay(target_path, original_filename=safe_filename)
+        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)
+        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)
+        task = convert_to_pdf.delay(target_path, original_filename=safe_filename, owner_id=upload_owner_id)
 
     # Check for exact duplicates (same SHA-256 hash) before returning.
     # This gives the caller an immediate warning without waiting for the pipeline.

app/config.py

@@ -116,6 +116,25 @@ class Settings(BaseSettings):
     session_secret: Optional[str] = None
     admin_group_name: str = "admin"
 
+    # Multi-user settings
+    multi_user_enabled: bool = Field(
+        default=False,
+        description=(
+            "Enable multi-user mode with individual document spaces per user. "
+            "When enabled, each authenticated user sees only their own documents, "
+            "uploads, and search results. Shared settings (AI, OCR) remain global. "
+            "Requires auth_enabled=True. Default: False (single-user/shared mode)."
+        ),
+    )
+    default_daily_upload_limit: int = Field(
+        default=0,
+        description=(
+            "Default maximum number of document uploads allowed per user per day "
+            "in multi-user mode. Set to 0 for unlimited. "
+            "Individual user limits can override this default. Default: 0 (unlimited)."
+        ),
+    )
+
     # Authentik
     authentik_client_id: Optional[str] = None
     authentik_client_secret: Optional[str] = None

app/models.py

@@ -24,6 +24,11 @@ class FileRecord(Base):
 
     id = Column(Integer, primary_key=True, index=True)
 
+    # Owner identifier for multi-user mode.
+    # Stores the user's unique identifier (e.g. email or OAuth sub claim).
+    # NULL means the file belongs to the shared/global space (single-user mode).
+    owner_id = Column(String, nullable=True, index=True)
+
     # Hash of the file content (e.g. SHA-256)
     # Note: duplicates are allowed so filehash is not unique
     filehash = Column(String, index=True, nullable=False)

app/tasks/convert_to_pdf.py

@@ -124,7 +124,9 @@ def _build_filename(file_path: str, original_filename: Optional[str], file_ext:
 
 
 @shared_task(bind=True)
-def convert_to_pdf(self, file_path: str, original_filename: Optional[str] = None) -> Optional[str]:
+def convert_to_pdf(
+    self, file_path: str, original_filename: Optional[str] = None, owner_id: Optional[str] = None
+) -> Optional[str]:
     """
     Converts a file to PDF using Gotenberg's API.
     Determines the appropriate Gotenberg endpoint based on the file's MIME type.
@@ -133,6 +135,7 @@ def convert_to_pdf(self, file_path: str, original_filename: Optional[str] = None
     Args:
         file_path: Path to the file to convert
         original_filename: Optional original filename (if different from path basename)
+        owner_id: Optional user identifier forwarded to process_document for multi-user mode.
     """
     task_id = self.request.id
     logger.info(f"[{task_id}] Starting PDF conversion: {file_path}")
@@ -332,9 +335,9 @@ def convert_to_pdf(self, file_path: str, original_filename: Optional[str] = None
                 # Change extension to .pdf for the original filename
                 original_base = os.path.splitext(original_filename)[0]
                 pdf_original_filename = f"{original_base}.pdf"
-                process_document.delay(converted_file_path, original_filename=pdf_original_filename)
+                process_document.delay(converted_file_path, original_filename=pdf_original_filename, owner_id=owner_id)
             else:
-                process_document.delay(converted_file_path)
+                process_document.delay(converted_file_path, owner_id=owner_id)
 
             return converted_file_path
         else:

app/tasks/process_document.py

@@ -25,7 +25,12 @@ logger = logging.getLogger(__name__)
 
 @celery.task(base=BaseTaskWithRetry, bind=True)
 def process_document(
-    self, original_local_file: str, original_filename: str = None, file_id: int = None, force_cloud_ocr: bool = False
+    self,
+    original_local_file: str,
+    original_filename: str = None,
+    file_id: int = None,
+    force_cloud_ocr: bool = False,
+    owner_id: str = None,
 ):
     """
     Process a document file and trigger appropriate text extraction.
@@ -37,6 +42,8 @@ def process_document(
                  detection and reuses the existing record (used for reprocessing).
         force_cloud_ocr: If True, forces Azure Document Intelligence OCR processing
                         regardless of embedded text quality. Used for re-processing.
+        owner_id: Optional user identifier for multi-user mode. When provided, the
+                  created FileRecord is associated with this user.
 
     Steps:
       1. Check if we have a FileRecord entry (via SHA-256 hash). If found, skip re-processing.
@@ -141,6 +148,7 @@ def process_document(
                     mime_type=mime_type,
                     is_duplicate=True,
                     duplicate_of_id=existing.id,
+                    owner_id=owner_id,
                 )
                 db.add(duplicate_record)
                 db.commit()
@@ -191,6 +199,7 @@ def process_document(
                 file_size=file_size,
                 mime_type=mime_type,
                 is_duplicate=False,
+                owner_id=owner_id,
             )
             db.add(new_record)
             db.commit()

app/utils/settings_service.py

@@ -78,6 +78,28 @@ SETTING_METADATA = {
         "required": False,
         "restart_required": True,
     },
+    "multi_user_enabled": {
+        "category": "Authentication",
+        "description": (
+            "Enable multi-user mode with individual document spaces per user. "
+            "Each user sees only their own documents, uploads, and search results. "
+            "Requires auth_enabled=True."
+        ),
+        "type": "boolean",
+        "sensitive": False,
+        "required": False,
+        "restart_required": True,
+    },
+    "default_daily_upload_limit": {
+        "category": "Authentication",
+        "description": (
+            "Default maximum document uploads allowed per user per day in multi-user mode. Set to 0 for unlimited."
+        ),
+        "type": "integer",
+        "sensitive": False,
+        "required": False,
+        "restart_required": False,
+    },
     "session_secret": {
         "category": "Authentication",
         "description": "Secret key for session encryption (min 32 characters)",

app/utils/user_scope.py

@@ -0,0 +1,72 @@
+"""
+User-scoping utilities for multi-user document isolation.
+
+When ``multi_user_enabled`` is ``True`` in settings, every document query
+is filtered by the authenticated user's identifier so that each user sees
+only their own documents.  When the flag is ``False`` (default), all
+documents are visible to all users (single-user / shared mode).
+"""
+
+import logging
+
+from fastapi import Request
+from sqlalchemy.orm import Query
+
+from app.models import FileRecord
+
+logger = logging.getLogger(__name__)
+
+
+def get_current_owner_id(request: Request) -> str | None:
+    """Extract the owner identifier for the current authenticated user.
+
+    The owner ID is derived from the user's session data.  It uses the
+    ``sub`` claim (OAuth subject) when available, falling back to
+    ``preferred_username`` or ``email``.  Returns ``None`` when no user
+    is authenticated.
+
+    Args:
+        request: The current FastAPI request with session data.
+
+    Returns:
+        A stable string identifier for the user, or ``None``.
+    """
+    user = request.session.get("user")
+    if not user or not isinstance(user, dict):
+        return None
+    # Prefer 'sub' (OAuth subject), then 'preferred_username', then 'email', then 'id'
+    return user.get("sub") or user.get("preferred_username") or user.get("email") or user.get("id")
+
+
+def apply_owner_filter(query: Query, request: Request) -> Query:
+    """Conditionally filter a ``FileRecord`` query by the current user.
+
+    When multi-user mode is enabled, only files whose ``owner_id``
+    matches the authenticated user are returned.  Admin users bypass
+    the filter and see all documents.
+
+    When multi-user mode is disabled the query is returned unchanged.
+
+    Args:
+        query: A SQLAlchemy query selecting ``FileRecord`` rows.
+        request: The current FastAPI request (for session inspection).
+
+    Returns:
+        The (possibly filtered) query.
+    """
+    from app.config import settings
+
+    if not settings.multi_user_enabled:
+        return query
+
+    user = request.session.get("user")
+    if isinstance(user, dict) and user.get("is_admin"):
+        # Admins see all documents in multi-user mode
+        return query
+
+    owner_id = get_current_owner_id(request)
+    if owner_id is None:
+        # No authenticated user — return empty result set
+        return query.filter(FileRecord.id < 0)
+
+    return query.filter(FileRecord.owner_id == owner_id)

migrations/versions/012_add_multi_user_support.py

@@ -0,0 +1,29 @@
+"""Add multi-user support: owner_id column on files table
+
+Revision ID: 012_add_multi_user_support
+Revises: 011_add_pdfa_paths
+Create Date: 2026-03-05
+
+"""
+
+from typing import Union
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "012_add_multi_user_support"
+down_revision: Union[str, None] = "011_add_pdfa_paths"
+depends_on: Union[str, None] = None
+
+
+def upgrade() -> None:
+    """Add owner_id column to files table for multi-user document isolation."""
+    op.add_column("files", sa.Column("owner_id", sa.String(), nullable=True))
+    op.create_index("ix_files_owner_id", "files", ["owner_id"])
+
+
+def downgrade() -> None:
+    """Remove owner_id column from files table."""
+    op.drop_index("ix_files_owner_id", table_name="files")
+    op.drop_column("files", "owner_id")