fix(tasks): dispatch per-user notifications on document processed/failed events

The per-user notification functions (notify_user_document_processed /
notify_user_document_failed) were defined but never called from the
document processing pipeline.

- Call notify_user_document_processed in finalize_document_storage
  when owner_id is available (creates in-app + email/webhook notifications)
- Add _dispatch_user_failure_notification helper to celery_app.py that
  extracts file_id from failed task args and dispatches
  notify_user_document_failed for document pipeline tasks
- Add comprehensive tests for both success and failure notification paths

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

app/celery_app.py

@@ -1,10 +1,14 @@
 # app/celery_app.py
 
+import logging
+
 from celery import Celery
 from celery.signals import task_failure, worker_ready
 
 from app.config import settings
 
+logger = logging.getLogger(__name__)
+
 celery = Celery(
     "document_processor",
     broker=settings.redis_url,
@@ -21,6 +25,62 @@ celery.conf.task_routes = {
     "app.tasks.*": {"queue": "document_processor"},
 }
 
+# Mapping of document pipeline task names to the positional index of ``file_id``
+# in their ``args`` tuple.  Tasks that always pass ``file_id`` as a keyword
+# argument (e.g. ``process_document``, ``finalize_document_storage``) are not
+# listed here — their ``file_id`` is found via ``kwargs`` instead.
+_FILE_ID_ARG_INDEX: dict[str, int] = {
+    "app.tasks.process_with_ocr.process_with_ocr": 1,
+    "app.tasks.extract_metadata_with_gpt.extract_metadata_with_gpt": 2,
+    "app.tasks.embed_metadata_into_pdf.embed_metadata_into_pdf": 3,
+}
+
+
+def _dispatch_user_failure_notification(sender, exception, args: list | None, kwargs: dict | None) -> None:
+    """Best-effort per-user failure notification for document pipeline tasks.
+
+    Extracts ``file_id`` from the failed task's arguments, looks up the owning
+    user from the database, and dispatches a ``document.failed`` notification.
+    """
+    from app.database import SessionLocal
+    from app.models import FileRecord
+    from app.utils.user_notification import notify_user_document_failed
+
+    task_name = sender.name if sender else ""
+    if not task_name.startswith("app.tasks."):
+        return
+
+    # 1. Resolve file_id from kwargs or positional args
+    file_id = (kwargs or {}).get("file_id")
+    if file_id is None:
+        idx = _FILE_ID_ARG_INDEX.get(task_name)
+        if idx is not None and args and len(args) > idx:
+            val = args[idx]
+            if isinstance(val, int):
+                file_id = val
+
+    if file_id is None:
+        return
+
+    # 2. Look up owner from the database
+    with SessionLocal() as db:
+        record = db.query(FileRecord).filter(FileRecord.id == file_id).first()
+        if not record or not record.owner_id:
+            return
+        owner_id = record.owner_id
+        filename = record.original_filename or record.local_filename or "unknown"
+
+    # 3. Dispatch per-user notification
+    import os
+
+    error_msg = f"{type(exception).__name__}: {exception}" if exception else "Unknown error"
+    notify_user_document_failed(
+        owner_id=owner_id,
+        filename=os.path.basename(filename),
+        error=error_msg,
+        file_id=file_id,
+    )
+
 
 @worker_ready.connect
 def init_sentry_on_worker_ready(**kwargs):
@@ -48,6 +108,10 @@ def task_failure_handler(
                 kwargs=kwargs or {},
             )
         except Exception as e:
-            import logging
+            logger.exception(f"Failed to send task failure notification: {e}")
 
-            logging.exception(f"Failed to send task failure notification: {e}")
+    # Also dispatch a per-user failure notification for document pipeline tasks
+    try:
+        _dispatch_user_failure_notification(sender, exception, args, kwargs)
+    except Exception:
+        logger.debug("Could not dispatch per-user failure notification", exc_info=True)

app/tasks/finalize_document_storage.py

@@ -21,8 +21,9 @@ from app.tasks.send_to_all import (
 # Import database and logging utils from main
 from app.utils import log_task_progress
 
-# Import notification utility
+# Import notification utilities
 from app.utils.notification import notify_file_processed
+from app.utils.user_notification import notify_user_document_processed
 
 logger = logging.getLogger(__name__)
 
@@ -139,4 +140,15 @@ def finalize_document_storage(self, original_file: str, processed_file: str, met
     except Exception as e:
         logger.warning(f"[WARNING] Failed to send file processed notification: {e}")
 
+    # 6. Send per-user notification
+    if owner_id:
+        try:
+            notify_user_document_processed(
+                owner_id=owner_id,
+                filename=os.path.basename(processed_file),
+                file_id=file_id,
+            )
+        except Exception as e:
+            logger.warning(f"[WARNING] Failed to send per-user processed notification: {e}")
+
     return {"status": "Completed", "file": processed_file}

tests/test_celery_app.py

@@ -224,3 +224,142 @@ class TestTaskFailureHandler:
         # Simply verify that importing the handler doesn't cause errors
         # The actual signal connection is tested implicitly by the other tests
         assert callable(task_failure_handler)
+
+
+@pytest.mark.unit
+class TestDispatchUserFailureNotification:
+    """Tests for _dispatch_user_failure_notification helper."""
+
+    @patch("app.celery_app._dispatch_user_failure_notification")
+    @patch("app.celery_app.settings")
+    @patch("app.utils.notification.notify_celery_failure")
+    def test_task_failure_handler_calls_user_failure_dispatch(self, mock_notify_sys, mock_settings, mock_dispatch):
+        """task_failure_handler also calls _dispatch_user_failure_notification."""
+        mock_settings.notify_on_task_failure = True
+
+        from app.celery_app import task_failure_handler
+
+        mock_sender = MagicMock()
+        mock_sender.name = "app.tasks.process_document.process_document"
+        exc = ValueError("OCR timeout")
+
+        task_failure_handler(
+            sender=mock_sender,
+            task_id="tid",
+            exception=exc,
+            args=["/tmp/f.pdf"],
+            kwargs={"file_id": 42},
+        )
+
+        mock_dispatch.assert_called_once_with(mock_sender, exc, ["/tmp/f.pdf"], {"file_id": 42})
+
+    def test_dispatch_ignores_non_document_tasks(self):
+        """Non app.tasks.* tasks should be silently ignored."""
+        from app.celery_app import _dispatch_user_failure_notification
+
+        sender = MagicMock()
+        sender.name = "celery.backend_cleanup"
+
+        # Should complete without error or DB access
+        _dispatch_user_failure_notification(sender, ValueError("x"), [], {})
+
+    def test_dispatch_ignores_when_no_file_id(self):
+        """If file_id is not in args or kwargs, nothing happens."""
+        from app.celery_app import _dispatch_user_failure_notification
+
+        sender = MagicMock()
+        sender.name = "app.tasks.process_document.process_document"
+
+        # No file_id anywhere
+        _dispatch_user_failure_notification(sender, ValueError("x"), ["/tmp/f.pdf"], {})
+
+    @patch("app.database.SessionLocal")
+    def test_dispatch_extracts_file_id_from_kwargs(self, mock_session):
+        """file_id should be extracted from kwargs when present."""
+        from app.celery_app import _dispatch_user_failure_notification
+
+        mock_db = MagicMock()
+        mock_record = MagicMock()
+        mock_record.owner_id = "alice@example.com"
+        mock_record.original_filename = "invoice.pdf"
+        mock_record.local_filename = "/tmp/invoice.pdf"
+        mock_db.query.return_value.filter.return_value.first.return_value = mock_record
+        mock_session.return_value.__enter__.return_value = mock_db
+
+        sender = MagicMock()
+        sender.name = "app.tasks.finalize_document_storage.finalize_document_storage"
+        exc = RuntimeError("Upload failed")
+
+        with patch("app.utils.user_notification.notify_user_document_failed") as mock_notify:
+            _dispatch_user_failure_notification(sender, exc, ["/tmp/f.pdf"], {"file_id": 10})
+
+        mock_notify.assert_called_once_with(
+            owner_id="alice@example.com",
+            filename="invoice.pdf",
+            error="RuntimeError: Upload failed",
+            file_id=10,
+        )
+
+    @patch("app.database.SessionLocal")
+    def test_dispatch_extracts_file_id_from_positional_args(self, mock_session):
+        """file_id should be extracted from positional args for known tasks."""
+        from app.celery_app import _dispatch_user_failure_notification
+
+        mock_db = MagicMock()
+        mock_record = MagicMock()
+        mock_record.owner_id = "bob@test.com"
+        mock_record.original_filename = "scan.pdf"
+        mock_record.local_filename = "/tmp/scan.pdf"
+        mock_db.query.return_value.filter.return_value.first.return_value = mock_record
+        mock_session.return_value.__enter__.return_value = mock_db
+
+        sender = MagicMock()
+        sender.name = "app.tasks.process_with_ocr.process_with_ocr"
+        exc = ValueError("OCR error")
+
+        with patch("app.utils.user_notification.notify_user_document_failed") as mock_notify:
+            # process_with_ocr: file_id is args[1]
+            _dispatch_user_failure_notification(sender, exc, ["filename.pdf", 77], {})
+
+        mock_notify.assert_called_once_with(
+            owner_id="bob@test.com",
+            filename="scan.pdf",
+            error="ValueError: OCR error",
+            file_id=77,
+        )
+
+    @patch("app.database.SessionLocal")
+    def test_dispatch_skips_when_no_owner(self, mock_session):
+        """When file record has no owner_id, no notification is sent."""
+        from app.celery_app import _dispatch_user_failure_notification
+
+        mock_db = MagicMock()
+        mock_record = MagicMock()
+        mock_record.owner_id = None
+        mock_db.query.return_value.filter.return_value.first.return_value = mock_record
+        mock_session.return_value.__enter__.return_value = mock_db
+
+        sender = MagicMock()
+        sender.name = "app.tasks.process_document.process_document"
+
+        with patch("app.utils.user_notification.notify_user_document_failed") as mock_notify:
+            _dispatch_user_failure_notification(sender, ValueError("x"), [], {"file_id": 5})
+
+        mock_notify.assert_not_called()
+
+    @patch("app.database.SessionLocal")
+    def test_dispatch_skips_when_record_not_found(self, mock_session):
+        """When file record doesn't exist, no notification is sent."""
+        from app.celery_app import _dispatch_user_failure_notification
+
+        mock_db = MagicMock()
+        mock_db.query.return_value.filter.return_value.first.return_value = None
+        mock_session.return_value.__enter__.return_value = mock_db
+
+        sender = MagicMock()
+        sender.name = "app.tasks.process_document.process_document"
+
+        with patch("app.utils.user_notification.notify_user_document_failed") as mock_notify:
+            _dispatch_user_failure_notification(sender, ValueError("x"), [], {"file_id": 999})
+
+        mock_notify.assert_not_called()

tests/test_finalize_storage.py

@@ -632,3 +632,137 @@ class TestFinalizeDocumentStorageUserRouting:
         mock_send_all.delay.assert_called_once_with("/workdir/processed/file.pdf", True, 400)
         mock_send_user.delay.assert_not_called()
         assert result["status"] == "Completed"
+
+
+@pytest.mark.unit
+class TestFinalizeDocumentStorageUserNotification:
+    """Tests for per-user notification dispatch in finalize_document_storage."""
+
+    @patch("app.tasks.finalize_document_storage.notify_user_document_processed")
+    @patch("app.tasks.finalize_document_storage.notify_file_processed")
+    @patch("app.tasks.finalize_document_storage.send_to_user_destinations")
+    @patch("app.tasks.finalize_document_storage.send_to_all_destinations")
+    @patch("app.tasks.finalize_document_storage.get_user_destination_count", return_value=2)
+    @patch("app.tasks.finalize_document_storage.get_configured_services_from_validator")
+    @patch("app.tasks.finalize_document_storage.log_task_progress")
+    @patch("app.tasks.finalize_document_storage.SessionLocal")
+    def test_dispatches_per_user_notification_when_owner_is_set(
+        self,
+        mock_session_local,
+        mock_log_progress,
+        mock_get_services,
+        mock_get_dest_count,
+        mock_send_all,
+        mock_send_user,
+        mock_notify_system,
+        mock_notify_user,
+    ):
+        """notify_user_document_processed is called when owner_id is available."""
+        mock_get_services.return_value = {"dropbox": True}
+
+        mock_db = MagicMock()
+        mock_session_local.return_value.__enter__.return_value = mock_db
+        mock_file_record = _make_file_record(100, owner_id="alice@example.com")
+        mock_db.query.return_value.filter.return_value.first.return_value = mock_file_record
+
+        with patch("app.tasks.finalize_document_storage.os.path.exists", return_value=True):
+            with patch("app.tasks.finalize_document_storage.os.path.getsize", return_value=1024):
+                with patch("app.tasks.finalize_document_storage.os.path.basename", return_value="doc.pdf"):
+                    finalize_document_storage.request.id = "test-task-id"
+
+                    finalize_document_storage.__wrapped__(
+                        original_file="/tmp/original.pdf",
+                        processed_file="/workdir/processed/doc.pdf",
+                        metadata={"filename": "doc.pdf"},
+                        file_id=100,
+                    )
+
+        mock_notify_user.assert_called_once_with(
+            owner_id="alice@example.com",
+            filename="doc.pdf",
+            file_id=100,
+        )
+
+    @patch("app.tasks.finalize_document_storage.notify_user_document_processed")
+    @patch("app.tasks.finalize_document_storage.notify_file_processed")
+    @patch("app.tasks.finalize_document_storage.send_to_user_destinations")
+    @patch("app.tasks.finalize_document_storage.send_to_all_destinations")
+    @patch("app.tasks.finalize_document_storage.get_user_destination_count", return_value=0)
+    @patch("app.tasks.finalize_document_storage.get_configured_services_from_validator")
+    @patch("app.tasks.finalize_document_storage.log_task_progress")
+    @patch("app.tasks.finalize_document_storage.SessionLocal")
+    def test_skips_per_user_notification_when_no_owner(
+        self,
+        mock_session_local,
+        mock_log_progress,
+        mock_get_services,
+        mock_get_dest_count,
+        mock_send_all,
+        mock_send_user,
+        mock_notify_system,
+        mock_notify_user,
+    ):
+        """notify_user_document_processed is NOT called when owner_id is None."""
+        mock_get_services.return_value = {"dropbox": True}
+
+        mock_db = MagicMock()
+        mock_session_local.return_value.__enter__.return_value = mock_db
+        mock_file_record = _make_file_record(200, owner_id=None)
+        mock_db.query.return_value.filter.return_value.first.return_value = mock_file_record
+
+        with patch("app.tasks.finalize_document_storage.os.path.exists", return_value=True):
+            with patch("app.tasks.finalize_document_storage.os.path.getsize", return_value=2048):
+                with patch("app.tasks.finalize_document_storage.os.path.basename", return_value="file.pdf"):
+                    finalize_document_storage.request.id = "test-task-id"
+
+                    finalize_document_storage.__wrapped__(
+                        original_file="/tmp/original.pdf",
+                        processed_file="/workdir/processed/file.pdf",
+                        metadata={"filename": "file.pdf"},
+                        file_id=200,
+                    )
+
+        mock_notify_user.assert_not_called()
+
+    @patch("app.tasks.finalize_document_storage.notify_user_document_processed")
+    @patch("app.tasks.finalize_document_storage.notify_file_processed")
+    @patch("app.tasks.finalize_document_storage.send_to_user_destinations")
+    @patch("app.tasks.finalize_document_storage.send_to_all_destinations")
+    @patch("app.tasks.finalize_document_storage.get_user_destination_count", return_value=0)
+    @patch("app.tasks.finalize_document_storage.get_configured_services_from_validator")
+    @patch("app.tasks.finalize_document_storage.log_task_progress")
+    @patch("app.tasks.finalize_document_storage.SessionLocal")
+    def test_per_user_notification_failure_does_not_break_task(
+        self,
+        mock_session_local,
+        mock_log_progress,
+        mock_get_services,
+        mock_get_dest_count,
+        mock_send_all,
+        mock_send_user,
+        mock_notify_system,
+        mock_notify_user,
+    ):
+        """Even if notify_user_document_processed raises, finalize returns success."""
+        mock_get_services.return_value = {"dropbox": True}
+        mock_notify_user.side_effect = RuntimeError("SMTP down")
+
+        mock_db = MagicMock()
+        mock_session_local.return_value.__enter__.return_value = mock_db
+        mock_file_record = _make_file_record(300, owner_id="bob@example.com")
+        mock_db.query.return_value.filter.return_value.first.return_value = mock_file_record
+
+        with patch("app.tasks.finalize_document_storage.os.path.exists", return_value=True):
+            with patch("app.tasks.finalize_document_storage.os.path.getsize", return_value=512):
+                with patch("app.tasks.finalize_document_storage.os.path.basename", return_value="a.pdf"):
+                    finalize_document_storage.request.id = "test-task-id"
+
+                    result = finalize_document_storage.__wrapped__(
+                        original_file="/tmp/original.pdf",
+                        processed_file="/workdir/processed/a.pdf",
+                        metadata={"filename": "a.pdf"},
+                        file_id=300,
+                    )
+
+        assert result["status"] == "Completed"
+        mock_notify_user.assert_called_once()