diff --git a/app/api/__init__.py b/app/api/__init__.py index b483248a..ec050431 100644 --- a/app/api/__init__.py +++ b/app/api/__init__.py @@ -17,6 +17,7 @@ from app.api.openai import router as openai_router from app.api.process import router as process_router from app.api.search import router as search_router from app.api.settings import router as settings_router +from app.api.queue import router as queue_router from app.api.url_upload import router as url_upload_router # Import all the individual routers @@ -42,3 +43,4 @@ router.include_router(logs_router) router.include_router(settings_router) router.include_router(url_upload_router) router.include_router(search_router) +router.include_router(queue_router) diff --git a/app/api/queue.py b/app/api/queue.py new file mode 100644 index 00000000..a2677678 --- /dev/null +++ b/app/api/queue.py @@ -0,0 +1,266 @@ +""" +Queue monitoring API endpoints. + +Provides endpoints to query Celery/Redis queue statistics and +database-level processing status for document pipeline visibility. +""" + +import logging +from typing import Any + +import redis +from fastapi import APIRouter, Depends +from sqlalchemy import func +from sqlalchemy.orm import Session + +from app.config import settings +from app.database import get_db +from app.models import FileProcessingStep, FileRecord + +logger = logging.getLogger(__name__) +router = APIRouter(prefix="/queue", tags=["queue"]) + + +def _get_redis_queue_length(redis_client: redis.Redis, queue_name: str) -> int: + """Get the number of messages in a Redis-backed Celery queue. + + Args: + redis_client: Connected Redis client instance. + queue_name: Name of the Celery queue to inspect. + + Returns: + Number of messages (tasks) waiting in the queue. + """ + try: + return redis_client.llen(queue_name) + except Exception: + logger.debug(f"Could not read queue length for '{queue_name}'") + return 0 + + +def _get_celery_inspect_stats() -> dict[str, Any]: + """Query the Celery inspect API for active, reserved, and scheduled tasks. + + Returns: + Dictionary with active, reserved, and scheduled task summaries. + """ + from app.celery_app import celery + + result: dict[str, Any] = { + "active": [], + "reserved": [], + "scheduled": [], + "workers_online": 0, + } + + try: + inspector = celery.control.inspect(timeout=2.0) + + active = inspector.active() or {} + reserved = inspector.reserved() or {} + scheduled = inspector.scheduled() or {} + + result["workers_online"] = len(active) + + for _worker, tasks in active.items(): + for task in tasks: + result["active"].append( + { + "id": task.get("id", ""), + "name": task.get("name", "unknown"), + "args": str(task.get("args", []))[:200], + "started": task.get("time_start"), + } + ) + + for _worker, tasks in reserved.items(): + for task in tasks: + result["reserved"].append( + { + "id": task.get("id", ""), + "name": task.get("name", "unknown"), + "args": str(task.get("args", []))[:200], + } + ) + + for _worker, tasks in scheduled.items(): + for task in tasks: + req = task.get("request", {}) + result["scheduled"].append( + { + "id": req.get("id", ""), + "name": req.get("name", "unknown"), + "eta": task.get("eta"), + } + ) + except Exception as exc: + logger.warning(f"Celery inspect failed (workers may be offline): {exc}") + + return result + + +def _get_db_processing_summary(db: Session) -> dict[str, Any]: + """Query the database for a summary of file processing states. + + Args: + db: SQLAlchemy database session. + + Returns: + Dictionary with counts of files by processing state. + """ + try: + total_files = db.query(func.count(FileRecord.id)).scalar() or 0 + + # Count files with at least one in_progress step + processing_count = ( + db.query(func.count(func.distinct(FileProcessingStep.file_id))) + .filter(FileProcessingStep.status == "in_progress") + .scalar() + or 0 + ) + + # Count files with at least one failure and no in_progress + failed_subq = ( + db.query(FileProcessingStep.file_id).filter(FileProcessingStep.status == "failure").distinct().subquery() + ) + in_progress_subq = ( + db.query(FileProcessingStep.file_id) + .filter(FileProcessingStep.status == "in_progress") + .distinct() + .subquery() + ) + failed_count = ( + db.query(func.count(func.distinct(failed_subq.c.file_id))) + .filter(~failed_subq.c.file_id.in_(db.query(in_progress_subq.c.file_id))) + .scalar() + or 0 + ) + + # Count files that have steps and all steps are success/skipped + all_step_files = db.query(FileProcessingStep.file_id).distinct().subquery() + # Files with any non-terminal step + non_terminal = ( + db.query(FileProcessingStep.file_id) + .filter(FileProcessingStep.status.in_(["in_progress", "pending", "failure"])) + .distinct() + .subquery() + ) + completed_count = ( + db.query(func.count(func.distinct(all_step_files.c.file_id))) + .filter(~all_step_files.c.file_id.in_(db.query(non_terminal.c.file_id))) + .scalar() + or 0 + ) + + # Files with no processing steps at all + files_with_steps = db.query(FileProcessingStep.file_id).distinct().subquery() + pending_count = ( + db.query(func.count(FileRecord.id)) + .filter(~FileRecord.id.in_(db.query(files_with_steps.c.file_id))) + .filter(FileRecord.is_duplicate.is_(False)) + .scalar() + or 0 + ) + + # Recent files being processed (last 20 in_progress or pending) + recent_processing = ( + db.query(FileRecord.id, FileRecord.original_filename, FileProcessingStep.step_name) + .join(FileProcessingStep, FileRecord.id == FileProcessingStep.file_id) + .filter(FileProcessingStep.status == "in_progress") + .order_by(FileProcessingStep.updated_at.desc()) + .limit(20) + .all() + ) + + recent_list = [ + {"file_id": r[0], "filename": r[1] or f"File #{r[0]}", "current_step": r[2]} for r in recent_processing + ] + + return { + "total_files": total_files, + "processing": processing_count, + "failed": failed_count, + "completed": completed_count, + "pending": pending_count, + "recent_processing": recent_list, + } + except Exception as exc: + logger.error(f"Error querying DB processing summary: {exc}") + return { + "total_files": 0, + "processing": 0, + "failed": 0, + "completed": 0, + "pending": 0, + "recent_processing": [], + } + + +@router.get("/stats") +def get_queue_stats(db: Session = Depends(get_db)) -> dict[str, Any]: + """Get comprehensive queue and processing statistics. + + Returns queue lengths from Redis, Celery worker inspection data, + and database-level processing summaries for the document pipeline. + + Returns: + Dictionary containing redis queue info, celery worker info, + and database processing summary. + """ + # 1. Redis queue lengths + queue_lengths: dict[str, int] = {} + try: + redis_client = redis.Redis.from_url(settings.redis_url, decode_responses=True) + for queue_name in ["document_processor", "default", "celery"]: + queue_lengths[queue_name] = _get_redis_queue_length(redis_client, queue_name) + redis_client.close() + except Exception as exc: + logger.warning(f"Could not connect to Redis: {exc}") + + total_queued = sum(queue_lengths.values()) + + # 2. Celery inspect + celery_stats = _get_celery_inspect_stats() + + # 3. DB summary + db_summary = _get_db_processing_summary(db) + + return { + "queues": queue_lengths, + "total_queued": total_queued, + "celery": celery_stats, + "db_summary": db_summary, + } + + +@router.get("/pending-count") +def get_pending_count(db: Session = Depends(get_db)) -> dict[str, int]: + """Get a lightweight count of queued + in-progress items for the files page banner. + + Returns: + Dictionary with total_pending count (queued in Redis + processing in DB). + """ + total_pending = 0 + + # Redis queue lengths + try: + redis_client = redis.Redis.from_url(settings.redis_url, decode_responses=True) + for queue_name in ["document_processor", "default", "celery"]: + total_pending += _get_redis_queue_length(redis_client, queue_name) + redis_client.close() + except Exception: + logger.debug("Could not connect to Redis for pending count") + + # DB in-progress count + try: + processing_count = ( + db.query(func.count(func.distinct(FileProcessingStep.file_id))) + .filter(FileProcessingStep.status == "in_progress") + .scalar() + or 0 + ) + total_pending += processing_count + except Exception: + logger.debug("Could not query DB for processing count") + + return {"total_pending": total_pending} diff --git a/app/views/__init__.py b/app/views/__init__.py index bbc00bab..ed55ff29 100644 --- a/app/views/__init__.py +++ b/app/views/__init__.py @@ -12,6 +12,7 @@ from app.views.general import router as general_router from app.views.google_drive import router as google_drive_router from app.views.license_routes import router as license_router # Add the license router from app.views.onedrive import router as onedrive_router +from app.views.queue import router as queue_router from app.views.search import router as search_router from app.views.settings import router as settings_router from app.views.status import router as status_router @@ -29,3 +30,4 @@ router.include_router(license_router) # Include the license router router.include_router(settings_router) router.include_router(filemanager_router) router.include_router(search_router) +router.include_router(queue_router) diff --git a/app/views/queue.py b/app/views/queue.py new file mode 100644 index 00000000..993fde4f --- /dev/null +++ b/app/views/queue.py @@ -0,0 +1,32 @@ +""" +Queue monitoring view for the admin dashboard. +""" + +import logging + +from fastapi import Depends, Request +from sqlalchemy.orm import Session + +from app.views.base import APIRouter, get_db, require_login, templates +from app.views.settings import require_admin_access + +logger = logging.getLogger(__name__) +router = APIRouter() + + +@router.get("/admin/queue") +@require_login +@require_admin_access +async def queue_dashboard(request: Request, db: Session = Depends(get_db)): + """ + Queue monitoring dashboard — admin only. + + Displays Celery/Redis queue statistics and database processing summaries + so administrators can monitor the document processing pipeline. + """ + return templates.TemplateResponse( + "queue_dashboard.html", + { + "request": request, + }, + ) diff --git a/frontend/templates/base.html b/frontend/templates/base.html index ebe09847..75dcb274 100644 --- a/frontend/templates/base.html +++ b/frontend/templates/base.html @@ -95,6 +95,9 @@ File Manager + + Queue Monitor + @@ -168,6 +171,9 @@ File Manager + + Queue Monitor + diff --git a/frontend/templates/files.html b/frontend/templates/files.html index cc4e62ae..9c9b13aa 100644 --- a/frontend/templates/files.html +++ b/frontend/templates/files.html @@ -370,6 +370,23 @@

File Records

+ + + {% if error %}

Error: {{ error }}

@@ -1050,5 +1067,35 @@ _searchCurrentPage = 1; } + + +
{% endblock %} diff --git a/frontend/templates/queue_dashboard.html b/frontend/templates/queue_dashboard.html new file mode 100644 index 00000000..20f7fdbe --- /dev/null +++ b/frontend/templates/queue_dashboard.html @@ -0,0 +1,319 @@ +{% extends "base.html" %} +{% block title %}Queue Monitor — DocuElevate{% endblock %} + +{% block head_extra %} + +{% endblock %} + +{% block content %} +
+ + +
+

+ Queue Monitor +

+

Real-time view of the document processing pipeline and Celery task queues.

+
+ + +
+ +

Loading queue statistics…

+
+ + + + + + +
+ + +{% endblock %} diff --git a/tests/test_queue_monitoring.py b/tests/test_queue_monitoring.py new file mode 100644 index 00000000..256948b1 --- /dev/null +++ b/tests/test_queue_monitoring.py @@ -0,0 +1,241 @@ +"""Tests for app/api/queue.py and app/views/queue.py modules.""" + +from unittest.mock import MagicMock, Mock, patch + +import pytest + + +@pytest.mark.unit +class TestGetRedisQueueLength: + """Tests for the _get_redis_queue_length helper.""" + + def test_returns_queue_length(self): + """Test returns queue length from Redis.""" + from app.api.queue import _get_redis_queue_length + + mock_redis = MagicMock() + mock_redis.llen.return_value = 42 + assert _get_redis_queue_length(mock_redis, "document_processor") == 42 + mock_redis.llen.assert_called_once_with("document_processor") + + def test_returns_zero_on_error(self): + """Test returns 0 when Redis call fails.""" + from app.api.queue import _get_redis_queue_length + + mock_redis = MagicMock() + mock_redis.llen.side_effect = Exception("Connection refused") + assert _get_redis_queue_length(mock_redis, "default") == 0 + + +@pytest.mark.unit +class TestGetCeleryInspectStats: + """Tests for the _get_celery_inspect_stats helper.""" + + @patch("app.celery_app.celery") + def test_returns_worker_stats(self, mock_celery_mod): + """Test returns active, reserved, scheduled tasks.""" + from app.api.queue import _get_celery_inspect_stats + + mock_inspector = MagicMock() + mock_inspector.active.return_value = { + "worker1": [ + {"id": "task-1", "name": "app.tasks.process_document.process_document", "args": [1], "time_start": 123} + ] + } + mock_inspector.reserved.return_value = { + "worker1": [{"id": "task-2", "name": "app.tasks.upload_to_s3.upload_to_s3", "args": [2]}] + } + mock_inspector.scheduled.return_value = { + "worker1": [{"request": {"id": "task-3", "name": "app.tasks.check_credentials.check_credentials"}, "eta": "2026-01-01"}] + } + mock_celery_mod.control.inspect.return_value = mock_inspector + + result = _get_celery_inspect_stats() + + assert result["workers_online"] == 1 + assert len(result["active"]) == 1 + assert result["active"][0]["id"] == "task-1" + assert len(result["reserved"]) == 1 + assert len(result["scheduled"]) == 1 + + @patch("app.celery_app.celery") + def test_handles_no_workers(self, mock_celery_mod): + """Test handles case where no workers are online.""" + from app.api.queue import _get_celery_inspect_stats + + mock_inspector = MagicMock() + mock_inspector.active.return_value = None + mock_inspector.reserved.return_value = None + mock_inspector.scheduled.return_value = None + mock_celery_mod.control.inspect.return_value = mock_inspector + + result = _get_celery_inspect_stats() + + assert result["workers_online"] == 0 + assert result["active"] == [] + assert result["reserved"] == [] + assert result["scheduled"] == [] + + @patch("app.celery_app.celery") + def test_handles_inspect_exception(self, mock_celery_mod): + """Test handles exception during inspect.""" + from app.api.queue import _get_celery_inspect_stats + + mock_celery_mod.control.inspect.side_effect = Exception("Broker unreachable") + + result = _get_celery_inspect_stats() + + assert result["workers_online"] == 0 + assert result["active"] == [] + + +@pytest.mark.unit +class TestGetDbProcessingSummary: + """Tests for the _get_db_processing_summary helper.""" + + def test_returns_summary(self, db_session): + """Test returns processing summary from DB.""" + from app.api.queue import _get_db_processing_summary + from app.models import FileProcessingStep, FileRecord + + # Create some test files + file1 = FileRecord(filehash="abc1", local_filename="f1.pdf", file_size=100, is_duplicate=False) + file2 = FileRecord(filehash="abc2", local_filename="f2.pdf", file_size=200, is_duplicate=False) + db_session.add_all([file1, file2]) + db_session.commit() + + # Add steps: file1 completed, file2 in_progress + step1 = FileProcessingStep(file_id=file1.id, step_name="extract_text", status="success") + step2 = FileProcessingStep(file_id=file2.id, step_name="extract_text", status="in_progress") + db_session.add_all([step1, step2]) + db_session.commit() + + result = _get_db_processing_summary(db_session) + + assert result["total_files"] == 2 + assert result["processing"] == 1 + assert isinstance(result["recent_processing"], list) + + def test_returns_empty_on_error(self): + """Test returns empty summary on DB error.""" + from app.api.queue import _get_db_processing_summary + + mock_db = MagicMock() + mock_db.query.side_effect = Exception("DB error") + + result = _get_db_processing_summary(mock_db) + + assert result["total_files"] == 0 + assert result["processing"] == 0 + assert result["recent_processing"] == [] + + +@pytest.mark.integration +class TestQueueStatsEndpoint: + """Tests for the GET /api/queue/stats endpoint.""" + + @patch("app.api.queue.redis.Redis") + @patch("app.celery_app.celery") + def test_queue_stats_returns_200(self, mock_celery_mod, mock_redis_cls, client): + """Test queue stats endpoint returns 200 with data.""" + # Mock Redis + mock_redis_instance = MagicMock() + mock_redis_instance.llen.return_value = 5 + mock_redis_cls.from_url.return_value = mock_redis_instance + + # Mock Celery inspector + mock_inspector = MagicMock() + mock_inspector.active.return_value = {} + mock_inspector.reserved.return_value = {} + mock_inspector.scheduled.return_value = {} + mock_celery_mod.control.inspect.return_value = mock_inspector + + response = client.get("/api/queue/stats") + assert response.status_code == 200 + + data = response.json() + assert "queues" in data + assert "total_queued" in data + assert "celery" in data + assert "db_summary" in data + + @patch("app.api.queue.redis.Redis") + def test_queue_stats_handles_redis_error(self, mock_redis_cls, client): + """Test queue stats handles Redis connection failure.""" + mock_redis_cls.from_url.side_effect = Exception("Redis unavailable") + + response = client.get("/api/queue/stats") + assert response.status_code == 200 + data = response.json() + assert data["total_queued"] == 0 + + +@pytest.mark.integration +class TestPendingCountEndpoint: + """Tests for the GET /api/queue/pending-count endpoint.""" + + @patch("app.api.queue.redis.Redis") + def test_pending_count_returns_200(self, mock_redis_cls, client): + """Test pending count endpoint returns 200.""" + mock_redis_instance = MagicMock() + mock_redis_instance.llen.return_value = 3 + mock_redis_cls.from_url.return_value = mock_redis_instance + + response = client.get("/api/queue/pending-count") + assert response.status_code == 200 + + data = response.json() + assert "total_pending" in data + assert data["total_pending"] >= 0 + + @patch("app.api.queue.redis.Redis") + def test_pending_count_redis_failure_still_works(self, mock_redis_cls, client): + """Test pending count still works when Redis is down.""" + mock_redis_cls.from_url.side_effect = Exception("Connection refused") + + response = client.get("/api/queue/pending-count") + assert response.status_code == 200 + data = response.json() + assert "total_pending" in data + + +@pytest.mark.integration +class TestQueueDashboardView: + """Tests for the GET /admin/queue view.""" + + def test_queue_dashboard_requires_login(self, client): + """Test queue dashboard requires authentication.""" + response = client.get("/admin/queue", follow_redirects=False) + assert response.status_code in [200, 302, 401] + + def test_queue_dashboard_non_admin_redirect(self, client): + """Test queue dashboard redirects non-admin users.""" + # Set non-admin session + with client: + client.cookies.set("session", "test") + response = client.get("/admin/queue", follow_redirects=False) + # Should redirect or deny since no admin session + assert response.status_code in [200, 302, 401] + + +@pytest.mark.unit +class TestQueueDashboardViewFunction: + """Tests for the queue_dashboard view function.""" + + @patch("app.views.queue.templates") + @pytest.mark.asyncio + async def test_queue_dashboard_returns_template(self, mock_templates): + """Test queue dashboard returns template response.""" + from app.views.queue import queue_dashboard + + mock_request = Mock() + mock_request.session = {"user": {"is_admin": True}} + mock_db = MagicMock() + + result = await queue_dashboard(mock_request, db=mock_db) + + mock_templates.TemplateResponse.assert_called_once() + call_args = mock_templates.TemplateResponse.call_args + assert call_args[0][0] == "queue_dashboard.html" + context = call_args[0][1] + assert "request" in context