From 195c3c344672357e09554996a1071e983fa92f8c Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Mon, 16 Mar 2026 16:10:01 +0000 Subject: [PATCH] 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 | 68 +++++++++++- app/tasks/finalize_document_storage.py | 14 ++- tests/test_celery_app.py | 139 +++++++++++++++++++++++++ tests/test_finalize_storage.py | 134 ++++++++++++++++++++++++ 4 files changed, 352 insertions(+), 3 deletions(-) diff --git a/app/celery_app.py b/app/celery_app.py index db10b6a4..59b34358 100644 --- a/app/celery_app.py +++ b/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) diff --git a/app/tasks/finalize_document_storage.py b/app/tasks/finalize_document_storage.py index e562564d..1b38f724 100644 --- a/app/tasks/finalize_document_storage.py +++ b/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} diff --git a/tests/test_celery_app.py b/tests/test_celery_app.py index af3014dc..7d98790b 100644 --- a/tests/test_celery_app.py +++ b/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() diff --git a/tests/test_finalize_storage.py b/tests/test_finalize_storage.py index 9b5c594e..214fc9d0 100644 --- a/tests/test_finalize_storage.py +++ b/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()