From 07d90758b8af64ecdeda633610a1f26fd2cc75d4 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Mon, 9 Mar 2026 17:52:43 +0000 Subject: [PATCH 1/4] Initial plan From c2d56d74754703e2f6cff71161f0bfd33eec20a8 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Mon, 9 Mar 2026 18:01:34 +0000 Subject: [PATCH 2/4] feat(tasks): add scheduled batch processing infrastructure (model, migration, tasks, API, view, template) Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com> --- app/api/scheduled_jobs.py | 258 ++++++++ app/celery_worker.py | 5 + app/models.py | 56 ++ app/tasks/batch_tasks.py | 325 ++++++++++ app/views/scheduled_jobs.py | 44 ++ frontend/templates/admin_scheduled_jobs.html | 558 ++++++++++++++++++ migrations/versions/026_add_scheduled_jobs.py | 51 ++ 7 files changed, 1297 insertions(+) create mode 100644 app/api/scheduled_jobs.py create mode 100644 app/tasks/batch_tasks.py create mode 100644 app/views/scheduled_jobs.py create mode 100644 frontend/templates/admin_scheduled_jobs.html create mode 100644 migrations/versions/026_add_scheduled_jobs.py diff --git a/app/api/scheduled_jobs.py b/app/api/scheduled_jobs.py new file mode 100644 index 00000000..7292d8d1 --- /dev/null +++ b/app/api/scheduled_jobs.py @@ -0,0 +1,258 @@ +""" +Admin API endpoints for managing scheduled batch processing jobs. + +All endpoints require admin privileges (checked via session ``is_admin`` flag). + +Available routes: + GET /api/admin/scheduled-jobs – list all scheduled jobs + PATCH /api/admin/scheduled-jobs/{id} – update schedule / enable-disable + POST /api/admin/scheduled-jobs/{id}/run-now – trigger a job immediately +""" + +import logging +from datetime import datetime, timezone +from typing import Annotated, Any + +from fastapi import APIRouter, Depends, HTTPException, Request, status +from pydantic import BaseModel, Field +from sqlalchemy.orm import Session + +from app.database import get_db +from app.models import ScheduledJob + +logger = logging.getLogger(__name__) +router = APIRouter(prefix="/admin/scheduled-jobs", tags=["admin-scheduled-jobs"]) + +DbSession = Annotated[Session, Depends(get_db)] + +# --------------------------------------------------------------------------- +# Authorisation helper +# --------------------------------------------------------------------------- + + +def _require_admin(request: Request) -> dict: + """Ensure the caller is an admin; raises HTTP 403 otherwise.""" + user = request.session.get("user") + if not user or not user.get("is_admin"): + raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required") + return user + + +AdminUser = Annotated[dict, Depends(_require_admin)] + + +# --------------------------------------------------------------------------- +# Pydantic schemas +# --------------------------------------------------------------------------- + + +class ScheduledJobResponse(BaseModel): + """Read model for a scheduled job.""" + + id: int + name: str + display_name: str + description: str | None + task_name: str + enabled: bool + schedule_type: str + cron_minute: str + cron_hour: str + cron_day_of_week: str + cron_day_of_month: str + cron_month_of_year: str + interval_seconds: int | None + last_run_at: datetime | None + last_run_status: str | None + last_run_detail: str | None + created_at: datetime | None + updated_at: datetime | None + + class Config: + from_attributes = True + + +class ScheduledJobUpdate(BaseModel): + """Writable fields for a scheduled job update (all optional).""" + + enabled: bool | None = Field(None, description="Whether the job is active") + schedule_type: str | None = Field(None, pattern="^(cron|interval)$", description="'cron' or 'interval'") + cron_minute: str | None = Field(None, max_length=50) + cron_hour: str | None = Field(None, max_length=50) + cron_day_of_week: str | None = Field(None, max_length=50) + cron_day_of_month: str | None = Field(None, max_length=50) + cron_month_of_year: str | None = Field(None, max_length=50) + interval_seconds: int | None = Field(None, ge=60, description="Interval in seconds (min 60)") + + +# --------------------------------------------------------------------------- +# Default job definitions – seeded into the DB on first startup +# --------------------------------------------------------------------------- + +DEFAULT_JOBS: list[dict[str, Any]] = [ + { + "name": "process-new-documents", + "display_name": "Process New Documents", + "description": ( + "Scans for documents that have been uploaded but never processed " + "and queues them through the full processing pipeline. " + "Runs hourly by default." + ), + "task_name": "app.tasks.batch_tasks.process_new_documents", + "enabled": True, + "schedule_type": "cron", + "cron_minute": "0", + "cron_hour": "*/1", + "cron_day_of_week": "*", + "cron_day_of_month": "*", + "cron_month_of_year": "*", + "interval_seconds": None, + }, + { + "name": "reprocess-failed-documents", + "display_name": "Reprocess Failed Documents", + "description": ( + "Finds documents whose last processing attempt failed and re-queues " + "them for reprocessing. Only picks up files that are not currently " + "being processed. Runs every 6 hours by default." + ), + "task_name": "app.tasks.batch_tasks.reprocess_failed_documents", + "enabled": True, + "schedule_type": "cron", + "cron_minute": "30", + "cron_hour": "*/6", + "cron_day_of_week": "*", + "cron_day_of_month": "*", + "cron_month_of_year": "*", + "interval_seconds": None, + }, + { + "name": "cleanup-temp-files", + "display_name": "Clean Up Temporary Files", + "description": ( + "Removes stale files from the workdir/tmp directory. " + "Only files older than 24 hours that are not referenced by any active " + "processing job are deleted. Runs daily at 03:30 UTC by default." + ), + "task_name": "app.tasks.batch_tasks.cleanup_temp_files", + "enabled": True, + "schedule_type": "cron", + "cron_minute": "30", + "cron_hour": "3", + "cron_day_of_week": "*", + "cron_day_of_month": "*", + "cron_month_of_year": "*", + "interval_seconds": None, + }, +] + + +def seed_default_scheduled_jobs(db: Session) -> None: + """ + Insert the built-in scheduled jobs if they do not already exist. + + Called from the FastAPI lifespan handler so the records are available + immediately after the first startup. + """ + for job_data in DEFAULT_JOBS: + existing = db.query(ScheduledJob).filter(ScheduledJob.name == job_data["name"]).first() + if existing is None: + db.add(ScheduledJob(**job_data)) + try: + db.commit() + except Exception as exc: + db.rollback() + logger.error("Failed to seed default scheduled jobs: %s", exc) + + +# --------------------------------------------------------------------------- +# Endpoints +# --------------------------------------------------------------------------- + + +@router.get("", response_model=list[ScheduledJobResponse]) +def list_scheduled_jobs(request: Request, db: DbSession, _admin: AdminUser) -> list[ScheduledJobResponse]: + """ + Return all scheduled jobs ordered by display name. + + Requires admin privileges. + """ + jobs = db.query(ScheduledJob).order_by(ScheduledJob.display_name).all() + return jobs # type: ignore[return-value] + + +@router.patch("/{job_id}", response_model=ScheduledJobResponse) +def update_scheduled_job( + job_id: int, + payload: ScheduledJobUpdate, + request: Request, + db: DbSession, + _admin: AdminUser, +) -> ScheduledJobResponse: + """ + Update schedule configuration or enabled state for a job. + + Only the fields included in the request body are modified. + Changes to the Celery Beat schedule take effect after the worker restarts. + + Requires admin privileges. + """ + job = db.query(ScheduledJob).filter(ScheduledJob.id == job_id).first() + if job is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Scheduled job not found") + + update_data = payload.model_dump(exclude_none=True) + if not update_data: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="No fields to update") + + for field, value in update_data.items(): + setattr(job, field, value) + + job.updated_at = datetime.now(timezone.utc) + + try: + db.commit() + db.refresh(job) + except Exception as exc: + db.rollback() + logger.error("Failed to update scheduled job %s: %s", job_id, exc) + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="Failed to update scheduled job", + ) from exc + + logger.info("Admin updated scheduled job %s (id=%s): %s", job.name, job_id, update_data) + return job # type: ignore[return-value] + + +@router.post("/{job_id}/run-now") +def run_scheduled_job_now( + job_id: int, + request: Request, + db: DbSession, + _admin: AdminUser, +) -> dict[str, Any]: + """ + Immediately dispatch the Celery task for the given scheduled job. + + The task is sent to the default queue; its result is tracked asynchronously + via the ``last_run_at`` / ``last_run_status`` fields updated by the task + itself. + + Requires admin privileges. + """ + job = db.query(ScheduledJob).filter(ScheduledJob.id == job_id).first() + if job is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Scheduled job not found") + + from app.celery_app import celery as celery_app + + task = celery_app.send_task(job.task_name) + logger.info("Admin triggered scheduled job %s (id=%s) manually, task_id=%s", job.name, job_id, task.id) + + return { + "status": "dispatched", + "job_id": job_id, + "job_name": job.name, + "task_id": task.id, + } diff --git a/app/celery_worker.py b/app/celery_worker.py index 378b71af..ca9d11f0 100644 --- a/app/celery_worker.py +++ b/app/celery_worker.py @@ -9,6 +9,11 @@ from app import tasks # noqa: F401 - Imports app/tasks.py so Celery can registe from app.celery_app import celery from app.config import settings from app.tasks.backup_tasks import cleanup_old_backups, create_backup # noqa: F401 +from app.tasks.batch_tasks import ( # noqa: F401 + cleanup_temp_files, + process_new_documents, + reprocess_failed_documents, +) from app.tasks.check_credentials import check_credentials from app.tasks.compute_embedding import backfill_missing_embeddings, compute_document_embedding # noqa: F401 from app.tasks.convert_to_pdf import convert_to_pdf # noqa: F401 diff --git a/app/models.py b/app/models.py index 4e5bd6ef..0cc7b53a 100644 --- a/app/models.py +++ b/app/models.py @@ -777,3 +777,59 @@ class InAppNotification(Base): is_read = Column(Boolean, nullable=False, default=False, index=True) file_id = Column(Integer, nullable=True) # Optional link to FileRecord created_at = Column(DateTime(timezone=True), server_default=func.now(), index=True) + + +class ScheduledJob(Base): + """ + Configuration record for an admin-managed scheduled batch processing job. + + Each row represents one recurring job entry. The Celery Beat schedule is + built from these rows at worker startup; changes take effect after the + worker process is restarted. + + Schedule types + -------------- + - ``"cron"`` – standard cron expression fields (minute/hour/…) + - ``"interval"`` – fixed interval in seconds (e.g. 3600 for hourly) + """ + + __tablename__ = "scheduled_jobs" + + id = Column(Integer, primary_key=True, index=True) + + # Unique machine-readable key used as the Celery Beat schedule entry name. + name = Column(String(100), unique=True, nullable=False, index=True) + + # Human-readable display name shown in the admin UI. + display_name = Column(String(255), nullable=False) + + # Short description of what the job does. + description = Column(Text, nullable=True) + + # Fully-qualified Celery task name, e.g. "app.tasks.batch_tasks.process_new_documents". + task_name = Column(String(255), nullable=False) + + # Whether the job is active. Inactive jobs are excluded from the beat schedule. + enabled = Column(Boolean, nullable=False, default=True) + + # Schedule type: "cron" or "interval". + schedule_type = Column(String(20), nullable=False, default="cron") + + # --- Cron fields (used when schedule_type == "cron") --- + cron_minute = Column(String(50), nullable=False, default="0") + cron_hour = Column(String(50), nullable=False, default="*") + cron_day_of_week = Column(String(50), nullable=False, default="*") + cron_day_of_month = Column(String(50), nullable=False, default="*") + cron_month_of_year = Column(String(50), nullable=False, default="*") + + # --- Interval field (used when schedule_type == "interval") --- + # Interval in seconds; e.g. 3600 = hourly, 86400 = daily. + interval_seconds = Column(Integer, nullable=True) + + # Timestamps populated by the worker after each execution. + last_run_at = Column(DateTime(timezone=True), nullable=True) + last_run_status = Column(String(20), nullable=True) # "success", "failed", "running" + last_run_detail = Column(Text, nullable=True) # Brief result summary or error + + created_at = Column(DateTime(timezone=True), server_default=func.now()) + updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) diff --git a/app/tasks/batch_tasks.py b/app/tasks/batch_tasks.py new file mode 100644 index 00000000..ae18d2c8 --- /dev/null +++ b/app/tasks/batch_tasks.py @@ -0,0 +1,325 @@ +""" +Scheduled batch processing tasks for DocuElevate. + +This module provides three Celery tasks that can be scheduled via Celery Beat +and managed through the admin UI (``/admin/scheduled-jobs``): + +- ``process_new_documents`` – Queue any documents that have never been processed. +- ``reprocess_failed_documents`` – Re-queue documents whose processing failed. +- ``cleanup_temp_files`` – Remove stale files from the ``workdir/tmp`` directory. + +Each task records its execution result back to the ``ScheduledJob`` table so +the admin UI can display last-run times and statuses. +""" + +import logging +import os +from datetime import datetime, timedelta, timezone +from pathlib import Path + +from app.celery_app import celery +from app.config import settings +from app.database import SessionLocal +from app.models import FileProcessingStep, FileRecord, ScheduledJob + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Internal helpers +# --------------------------------------------------------------------------- + +_PROCESSING_STEPS = { + "create_file_record", + "check_text", + "extract_text", + "process_with_ocr", + "process_with_azure_document_intelligence", + "extract_metadata_with_gpt", + "embed_metadata_into_pdf", + "finalize_document_storage", + "send_to_all_destinations", +} + + +def _update_job_status(job_name: str, status: str, detail: str) -> None: + """Persist run status back to the ScheduledJob row for display in the UI.""" + try: + with SessionLocal() as db: + job = db.query(ScheduledJob).filter(ScheduledJob.name == job_name).first() + if job: + job.last_run_at = datetime.now(timezone.utc) + job.last_run_status = status + job.last_run_detail = detail + db.commit() + except Exception as exc: # pragma: no cover – best-effort status update + logger.warning("Could not update ScheduledJob status for %s: %s", job_name, exc) + + +# --------------------------------------------------------------------------- +# Task: process new (unprocessed) documents +# --------------------------------------------------------------------------- + + +@celery.task(name="app.tasks.batch_tasks.process_new_documents") +def process_new_documents() -> dict: + """ + Queue all documents that have never been processed. + + A document is considered *new* when it has no ``FileProcessingStep`` rows + that match the core pipeline steps. The task loads each qualifying + ``FileRecord``, verifies that the original file still exists on disk, and + dispatches ``process_document`` for each one. + + Returns a summary dict with ``queued`` and ``skipped`` counts. + """ + from app.tasks.process_document import process_document # avoid circular import + + job_name = "process-new-documents" + logger.info("[batch] Starting process_new_documents task") + + try: + with SessionLocal() as db: + # Files that already have at least one processing step recorded. + processed_file_ids = ( + db.query(FileProcessingStep.file_id) + .filter(FileProcessingStep.step_name.in_(_PROCESSING_STEPS)) + .distinct() + .subquery() + ) + + # Candidate files: non-duplicate records with no processing steps yet. + candidates = ( + db.query(FileRecord) + .filter(FileRecord.is_duplicate.is_(False)) + .filter(~FileRecord.id.in_(db.query(processed_file_ids.c.file_id))) + .all() + ) + + queued = 0 + skipped = 0 + for record in candidates: + if not record.local_filename or not os.path.exists(record.local_filename): + logger.warning( + "[batch] Skipping file_id=%s — local file not found: %s", + record.id, + record.local_filename, + ) + skipped += 1 + continue + process_document.delay( + record.local_filename, + original_filename=record.original_filename, + file_id=record.id, + owner_id=record.owner_id, + ) + queued += 1 + + detail = f"Queued {queued} document(s) for processing; skipped {skipped} (file not on disk)." + logger.info("[batch] process_new_documents: %s", detail) + _update_job_status(job_name, "success", detail) + return {"queued": queued, "skipped": skipped} + + except Exception as exc: + detail = f"Error: {exc}" + logger.error("[batch] process_new_documents failed: %s", exc, exc_info=True) + _update_job_status(job_name, "failed", detail) + return {"queued": 0, "skipped": 0, "error": str(exc)} + + +# --------------------------------------------------------------------------- +# Task: reprocess failed documents +# --------------------------------------------------------------------------- + + +@celery.task(name="app.tasks.batch_tasks.reprocess_failed_documents") +def reprocess_failed_documents() -> dict: + """ + Re-queue documents whose most-recent processing attempt failed. + + Only files that have at least one ``FileProcessingStep`` with + ``status == "failure"`` **and** no currently ``in_progress`` steps are + selected so that actively-running jobs are not interrupted. + + Returns a summary dict with ``queued`` and ``skipped`` counts. + """ + from app.tasks.process_document import process_document # avoid circular import + + job_name = "reprocess-failed-documents" + logger.info("[batch] Starting reprocess_failed_documents task") + + try: + with SessionLocal() as db: + # Files with at least one failed step. + failed_file_ids = ( + db.query(FileProcessingStep.file_id) + .filter(FileProcessingStep.step_name.in_(_PROCESSING_STEPS)) + .filter(FileProcessingStep.status == "failure") + .distinct() + .subquery() + ) + + # Exclude files that are currently being processed. + in_progress_file_ids = ( + db.query(FileProcessingStep.file_id) + .filter(FileProcessingStep.status == "in_progress") + .distinct() + .subquery() + ) + + candidates = ( + db.query(FileRecord) + .filter(FileRecord.is_duplicate.is_(False)) + .filter(FileRecord.id.in_(db.query(failed_file_ids.c.file_id))) + .filter(~FileRecord.id.in_(db.query(in_progress_file_ids.c.file_id))) + .all() + ) + + queued = 0 + skipped = 0 + for record in candidates: + if not record.local_filename or not os.path.exists(record.local_filename): + logger.warning( + "[batch] Skipping file_id=%s — local file not found: %s", + record.id, + record.local_filename, + ) + skipped += 1 + continue + process_document.delay( + record.local_filename, + original_filename=record.original_filename, + file_id=record.id, + owner_id=record.owner_id, + ) + queued += 1 + + detail = f"Re-queued {queued} failed document(s); skipped {skipped} (file not on disk)." + logger.info("[batch] reprocess_failed_documents: %s", detail) + _update_job_status(job_name, "success", detail) + return {"queued": queued, "skipped": skipped} + + except Exception as exc: + detail = f"Error: {exc}" + logger.error("[batch] reprocess_failed_documents failed: %s", exc, exc_info=True) + _update_job_status(job_name, "failed", detail) + return {"queued": 0, "skipped": 0, "error": str(exc)} + + +# --------------------------------------------------------------------------- +# Task: clean up temporary files +# --------------------------------------------------------------------------- + +#: Files in ``workdir/tmp`` that are older than this threshold are deleted. +_TEMP_FILE_MAX_AGE_HOURS: int = 24 + + +@celery.task(name="app.tasks.batch_tasks.cleanup_temp_files") +def cleanup_temp_files(max_age_hours: int = _TEMP_FILE_MAX_AGE_HOURS) -> dict: + """ + Delete stale files from the ``workdir/tmp`` directory. + + A file is considered stale when **both** of the following are true: + + 1. Its modification time is older than *max_age_hours* (default 24 h). + 2. No ``FileRecord.local_filename`` points to it **or** the file is not + referenced by any active in-progress processing step. + + This prevents accidental deletion of files that are being actively + processed by the pipeline. + + Args: + max_age_hours: Minimum age (in hours) before a temp file is eligible + for deletion. Defaults to 24. + + Returns: + A summary dict with ``deleted`` and ``skipped`` counts. + """ + job_name = "cleanup-temp-files" + logger.info("[batch] Starting cleanup_temp_files (max_age_hours=%s)", max_age_hours) + + tmp_dir = Path(settings.workdir) / "tmp" + if not tmp_dir.exists(): + detail = "workdir/tmp does not exist; nothing to clean." + logger.info("[batch] cleanup_temp_files: %s", detail) + _update_job_status(job_name, "success", detail) + return {"deleted": 0, "skipped": 0} + + cutoff = datetime.now(timezone.utc) - timedelta(hours=max_age_hours) + + deleted = 0 + skipped = 0 + errors = 0 + + try: + with SessionLocal() as db: + # Collect filenames actively referenced by in-progress processing steps. + in_progress_filenames: set[str] = set() + in_progress_records = ( + db.query(FileRecord.local_filename) + .join(FileProcessingStep, FileProcessingStep.file_id == FileRecord.id) + .filter(FileProcessingStep.status == "in_progress") + .distinct() + .all() + ) + for row in in_progress_records: + if row.local_filename: + in_progress_filenames.add(os.path.basename(row.local_filename)) + + # Also collect all filenames referenced by FileRecord.local_filename + # that point into workdir/tmp (files still in the tmp pipeline). + active_tmp_filenames: set[str] = set() + tmp_dir_str = str(tmp_dir.resolve()) + active_records = ( + db.query(FileRecord.local_filename) + .filter(FileRecord.local_filename.like(f"{tmp_dir_str}%")) + .all() + ) + for row in active_records: + if row.local_filename: + active_tmp_filenames.add(os.path.basename(row.local_filename)) + + protected_basenames = in_progress_filenames | active_tmp_filenames + + for entry in tmp_dir.iterdir(): + if not entry.is_file(): + continue + + # Check modification time. + try: + mtime = datetime.fromtimestamp(entry.stat().st_mtime, tz=timezone.utc) + except OSError: + skipped += 1 + continue + + if mtime >= cutoff: + skipped += 1 + continue + + if entry.name in protected_basenames: + logger.debug("[batch] cleanup_temp_files: keeping protected file %s", entry.name) + skipped += 1 + continue + + try: + entry.unlink() + logger.debug("[batch] cleanup_temp_files: deleted %s", entry) + deleted += 1 + except OSError as exc: + logger.warning("[batch] cleanup_temp_files: could not delete %s: %s", entry, exc) + errors += 1 + + detail = ( + f"Deleted {deleted} stale temp file(s); " + f"skipped {skipped} (too new or protected); " + f"{errors} error(s)." + ) + status = "failed" if errors and not deleted else "success" + logger.info("[batch] cleanup_temp_files: %s", detail) + _update_job_status(job_name, status, detail) + return {"deleted": deleted, "skipped": skipped, "errors": errors} + + except Exception as exc: + detail = f"Error: {exc}" + logger.error("[batch] cleanup_temp_files failed: %s", exc, exc_info=True) + _update_job_status(job_name, "failed", detail) + return {"deleted": 0, "skipped": 0, "errors": 1, "error": str(exc)} diff --git a/app/views/scheduled_jobs.py b/app/views/scheduled_jobs.py new file mode 100644 index 00000000..a0d68bf8 --- /dev/null +++ b/app/views/scheduled_jobs.py @@ -0,0 +1,44 @@ +"""Admin view: scheduled batch processing jobs management page.""" + +import logging + +from fastapi import HTTPException, Request, status +from fastapi.responses import RedirectResponse + +from app.views.base import APIRouter, require_login, settings, templates + +logger = logging.getLogger(__name__) +router = APIRouter() + + +def _require_admin(request: Request): + """Return the session user if they are an admin, else None.""" + user = request.session.get("user") + if not user or not user.get("is_admin"): + logger.warning("Non-admin user attempted to access /admin/scheduled-jobs") + return None + return user + + +@router.get("/admin/scheduled-jobs") +@require_login +async def scheduled_jobs_page(request: Request): + """Admin scheduled jobs management page.""" + user = _require_admin(request) + if user is None: + return RedirectResponse(url="/", status_code=status.HTTP_302_FOUND) + + try: + return templates.TemplateResponse( + "admin_scheduled_jobs.html", + { + "request": request, + "app_version": settings.version, + }, + ) + except Exception as e: + logger.error(f"Error loading scheduled jobs page: {e}") + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="Failed to load scheduled jobs page", + ) diff --git a/frontend/templates/admin_scheduled_jobs.html b/frontend/templates/admin_scheduled_jobs.html new file mode 100644 index 00000000..cf4ed93d --- /dev/null +++ b/frontend/templates/admin_scheduled_jobs.html @@ -0,0 +1,558 @@ +{% extends "base.html" %} +{% block title %}Scheduled Jobs – Admin – DocuElevate{% endblock %} + +{% block content %} +
+ + +
+
+

+ + Scheduled Jobs + Admin Only +

+

+ Manage and trigger scheduled batch processing jobs. Schedule changes take effect after the worker restarts. +

+
+ +
+ + + + + +
+ +
+ How scheduling works: + These jobs are executed by the Celery Beat scheduler running inside the worker container. + Schedule changes (cron / interval) and enable / disable toggles are persisted immediately, + but the Celery Beat process reads the schedule only at startup — so changes take effect + after the worker is restarted. You can always trigger any job right now + using the Run Now button without restarting. +
+
+ + +
+
+ Loading… +
+
+ + +
+ + + +
+ + + + +
+{% endblock %} + +{% block scripts %} + +{% endblock %} diff --git a/migrations/versions/026_add_scheduled_jobs.py b/migrations/versions/026_add_scheduled_jobs.py new file mode 100644 index 00000000..a8e032b8 --- /dev/null +++ b/migrations/versions/026_add_scheduled_jobs.py @@ -0,0 +1,51 @@ +"""Add scheduled_jobs table for admin-managed batch processing schedules. + +Revision ID: 026_add_scheduled_jobs +Revises: 025_add_user_notifications +Create Date: 2026-03-09 +""" + +from typing import Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "026_add_scheduled_jobs" +down_revision: Union[str, None] = "025_add_user_notifications" +depends_on: Union[str, None] = None + + +def upgrade() -> None: + """Create scheduled_jobs table.""" + op.create_table( + "scheduled_jobs", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("name", sa.String(100), nullable=False), + sa.Column("display_name", sa.String(255), nullable=False), + sa.Column("description", sa.Text(), nullable=True), + sa.Column("task_name", sa.String(255), nullable=False), + sa.Column("enabled", sa.Boolean(), nullable=False, server_default="1"), + sa.Column("schedule_type", sa.String(20), nullable=False, server_default="cron"), + sa.Column("cron_minute", sa.String(50), nullable=False, server_default="0"), + sa.Column("cron_hour", sa.String(50), nullable=False, server_default="*"), + sa.Column("cron_day_of_week", sa.String(50), nullable=False, server_default="*"), + sa.Column("cron_day_of_month", sa.String(50), nullable=False, server_default="*"), + sa.Column("cron_month_of_year", sa.String(50), nullable=False, server_default="*"), + sa.Column("interval_seconds", sa.Integer(), nullable=True), + sa.Column("last_run_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("last_run_status", sa.String(20), nullable=True), + sa.Column("last_run_detail", sa.Text(), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("name", name="uq_scheduled_jobs_name"), + ) + op.create_index("ix_scheduled_jobs_id", "scheduled_jobs", ["id"]) + op.create_index("ix_scheduled_jobs_name", "scheduled_jobs", ["name"]) + + +def downgrade() -> None: + """Drop scheduled_jobs table.""" + op.drop_index("ix_scheduled_jobs_name", "scheduled_jobs") + op.drop_index("ix_scheduled_jobs_id", "scheduled_jobs") + op.drop_table("scheduled_jobs") From 5bc1fac79cd1f4079df76e565c366c461457cb1a Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Mon, 9 Mar 2026 18:25:26 +0000 Subject: [PATCH 3/4] feat(tasks): add 5 additional scheduled jobs (expire_shared_links, prune_processing_logs, prune_old_notifications, backfill_missing_metadata, sync_search_index) Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com> --- app/api/__init__.py | 2 + app/api/scheduled_jobs.py | 93 +++ app/celery_worker.py | 72 +++ app/main.py | 14 + app/tasks/batch_tasks.py | 368 +++++++++++- app/views/__init__.py | 2 + docs/ScheduledJobs.md | 226 ++++++++ frontend/templates/base.html | 6 + tests/conftest.py | 1 + tests/test_scheduled_jobs.py | 1040 ++++++++++++++++++++++++++++++++++ 10 files changed, 1812 insertions(+), 12 deletions(-) create mode 100644 docs/ScheduledJobs.md create mode 100644 tests/test_scheduled_jobs.py diff --git a/app/api/__init__.py b/app/api/__init__.py index 0b083a9b..ae98cbd7 100644 --- a/app/api/__init__.py +++ b/app/api/__init__.py @@ -29,6 +29,7 @@ from app.api.plans import router as plans_router from app.api.process import router as process_router from app.api.queue import router as queue_router from app.api.saved_searches import router as saved_searches_router +from app.api.scheduled_jobs import router as scheduled_jobs_router from app.api.search import router as search_router from app.api.settings import router as settings_router from app.api.shared_links import public_router as shared_links_public_router @@ -80,3 +81,4 @@ router.include_router(pipelines_router) router.include_router(imap_accounts_router) router.include_router(integrations_router) router.include_router(notifications_router) +router.include_router(scheduled_jobs_router) diff --git a/app/api/scheduled_jobs.py b/app/api/scheduled_jobs.py index 7292d8d1..56f8b62e 100644 --- a/app/api/scheduled_jobs.py +++ b/app/api/scheduled_jobs.py @@ -144,6 +144,99 @@ DEFAULT_JOBS: list[dict[str, Any]] = [ "cron_month_of_year": "*", "interval_seconds": None, }, + { + "name": "expire-shared-links", + "display_name": "Expire Stale Shared Links", + "description": ( + "Marks shared document links as inactive when their expiry time has passed. " + "Access is already blocked at request time, but this task keeps the " + "management UI counts accurate. Runs daily at 01:00 UTC by default." + ), + "task_name": "app.tasks.batch_tasks.expire_shared_links", + "enabled": True, + "schedule_type": "cron", + "cron_minute": "0", + "cron_hour": "1", + "cron_day_of_week": "*", + "cron_day_of_month": "*", + "cron_month_of_year": "*", + "interval_seconds": None, + }, + { + "name": "prune-processing-logs", + "display_name": "Prune Old Processing Logs", + "description": ( + "Deletes processing log entries and settings audit log entries older than " + "30 days to prevent unbounded database growth. " + "Runs weekly on Sunday at 04:00 UTC by default." + ), + "task_name": "app.tasks.batch_tasks.prune_processing_logs", + "enabled": True, + "schedule_type": "cron", + "cron_minute": "0", + "cron_hour": "4", + "cron_day_of_week": "0", + "cron_day_of_month": "*", + "cron_month_of_year": "*", + "interval_seconds": None, + }, + { + "name": "prune-old-notifications", + "display_name": "Prune Old Notifications", + "description": ( + "Deletes read in-app notifications older than 30 days. " + "Unread notifications are never deleted. " + "Runs weekly on Sunday at 04:30 UTC by default." + ), + "task_name": "app.tasks.batch_tasks.prune_old_notifications", + "enabled": True, + "schedule_type": "cron", + "cron_minute": "30", + "cron_hour": "4", + "cron_day_of_week": "0", + "cron_day_of_month": "*", + "cron_month_of_year": "*", + "interval_seconds": None, + }, + { + "name": "backfill-missing-metadata", + "display_name": "Backfill Missing AI Metadata", + "description": ( + "Re-triggers AI metadata extraction for documents that have extracted " + "text but no AI metadata yet (e.g., processed before an AI provider " + "was configured). Processes up to 50 documents per run. " + "Runs every 6 hours by default." + ), + "task_name": "app.tasks.batch_tasks.backfill_missing_metadata", + "enabled": True, + "schedule_type": "cron", + "cron_minute": "0", + "cron_hour": "*/6", + "cron_day_of_week": "*", + "cron_day_of_month": "*", + "cron_month_of_year": "*", + "interval_seconds": None, + }, + { + "name": "sync-search-index", + "display_name": "Sync Search Index", + "description": ( + "Indexes documents that have OCR text or AI metadata but are missing " + "from the Meilisearch search index. Useful after enabling search on " + "an existing installation or after an index rebuild. " + "Processes up to 100 documents per run. " + "Runs hourly by default." + ), + "task_name": "app.tasks.batch_tasks.sync_search_index", + "enabled": True, + "schedule_type": "cron", + "cron_minute": "15", + "cron_hour": "*/1", + "cron_day_of_week": "*", + "cron_day_of_month": "*", + "cron_month_of_year": "*", + "interval_seconds": None, + }, ] diff --git a/app/celery_worker.py b/app/celery_worker.py index ca9d11f0..4881f8a9 100644 --- a/app/celery_worker.py +++ b/app/celery_worker.py @@ -1,5 +1,7 @@ #!/usr/bin/env python3 +import logging + from celery.schedules import crontab # Ensure tasks are loaded @@ -10,9 +12,14 @@ from app.celery_app import celery from app.config import settings from app.tasks.backup_tasks import cleanup_old_backups, create_backup # noqa: F401 from app.tasks.batch_tasks import ( # noqa: F401 + backfill_missing_metadata, cleanup_temp_files, + expire_shared_links, process_new_documents, + prune_old_notifications, + prune_processing_logs, reprocess_failed_documents, + sync_search_index, ) from app.tasks.check_credentials import check_credentials from app.tasks.compute_embedding import backfill_missing_embeddings, compute_document_embedding # noqa: F401 @@ -53,6 +60,8 @@ from app.tasks.webhook_tasks import deliver_webhook_task # noqa: F401 # Register the settings reload signal handler so workers pick up config changes from app.utils.settings_sync import register_settings_reload_signal +logger = logging.getLogger(__name__) + register_settings_reload_signal() celery.conf.task_routes = { @@ -164,3 +173,66 @@ celery.conf.beat_schedule = { # Remove None entries from beat_schedule celery.conf.beat_schedule = {k: v for k, v in celery.conf.beat_schedule.items() if v is not None} + +# --------------------------------------------------------------------------- +# Load admin-managed scheduled jobs from the database +# --------------------------------------------------------------------------- +# These jobs are defined in the ``scheduled_jobs`` table (seeded by +# ``app.api.scheduled_jobs.seed_default_scheduled_jobs``) and can be +# enabled/disabled and rescheduled via the admin UI at /admin/scheduled-jobs. +# The schedule is read once at worker startup; changes take effect after +# the worker is restarted. + + +def _load_db_scheduled_jobs() -> None: + """ + Extend ``celery.conf.beat_schedule`` with entries from the ``scheduled_jobs`` + database table. + + Only rows with ``enabled=True`` are added. Rows whose ``name`` key + already exists in the static schedule (defined above) are skipped so + that hardcoded entries cannot be overridden accidentally. + + Failures are logged as warnings and do not prevent the worker from + starting. + """ + try: + from app.database import SessionLocal + from app.models import ScheduledJob + + with SessionLocal() as db: + jobs = db.query(ScheduledJob).filter(ScheduledJob.enabled.is_(True)).all() + + added = 0 + for job in jobs: + if job.name in celery.conf.beat_schedule: + # Static entry takes precedence; skip silently. + continue + + if job.schedule_type == "interval" and job.interval_seconds: + from celery.schedules import schedule as interval_schedule + + sched = interval_schedule(run_every=job.interval_seconds) + else: + # Default to cron. + sched = crontab( + minute=job.cron_minute, + hour=job.cron_hour, + day_of_week=job.cron_day_of_week, + day_of_month=job.cron_day_of_month, + month_of_year=job.cron_month_of_year, + ) + + celery.conf.beat_schedule[job.name] = { + "task": job.task_name, + "schedule": sched, + "options": {"expires": 3600}, + } + added += 1 + + logger.info("Loaded %d scheduled job(s) from database into Celery Beat.", added) + except Exception as exc: + logger.warning("Could not load scheduled jobs from database: %s", exc) + + +_load_db_scheduled_jobs() diff --git a/app/main.py b/app/main.py index 11cb9410..228e8ff0 100644 --- a/app/main.py +++ b/app/main.py @@ -132,6 +132,20 @@ async def lifespan(app: FastAPI): except Exception: logging.debug("Default pipeline seeding skipped — DB may not be ready yet") # noqa: S110 + # Seed the default scheduled batch processing jobs so they appear in the + # admin UI (/admin/scheduled-jobs) on first startup. + try: + from app.api.scheduled_jobs import seed_default_scheduled_jobs as _seed_jobs + from app.database import SessionLocal as _SessionLocal # noqa: F811 + + _db_jobs = _SessionLocal() + try: + _seed_jobs(_db_jobs) + finally: + _db_jobs.close() + except Exception: + logging.debug("Scheduled jobs seeding skipped — DB may not be ready yet") # noqa: S110 + # Application is now running yield diff --git a/app/tasks/batch_tasks.py b/app/tasks/batch_tasks.py index ae18d2c8..006b469b 100644 --- a/app/tasks/batch_tasks.py +++ b/app/tasks/batch_tasks.py @@ -1,17 +1,31 @@ """ Scheduled batch processing tasks for DocuElevate. -This module provides three Celery tasks that can be scheduled via Celery Beat +This module provides Celery tasks that can be scheduled via Celery Beat and managed through the admin UI (``/admin/scheduled-jobs``): -- ``process_new_documents`` – Queue any documents that have never been processed. +Core batch jobs +--------------- +- ``process_new_documents`` – Queue documents that have never been processed. - ``reprocess_failed_documents`` – Re-queue documents whose processing failed. -- ``cleanup_temp_files`` – Remove stale files from the ``workdir/tmp`` directory. +- ``cleanup_temp_files`` – Remove stale files from the ``workdir/tmp`` directory. + +Maintenance / housekeeping jobs +-------------------------------- +- ``expire_shared_links`` – Auto-revoke SharedLinks whose ``expires_at`` has passed. +- ``prune_processing_logs`` – Delete old rows from ``processing_logs`` and + ``settings_audit_log`` to prevent unbounded table growth. +- ``prune_old_notifications`` – Delete old read ``in_app_notifications`` rows. +- ``backfill_missing_metadata`` – Re-trigger AI metadata extraction for completed files + that have OCR text but no ``ai_metadata``. +- ``sync_search_index`` – Index documents in Meilisearch that have OCR text / + metadata but are not yet in the search index. Each task records its execution result back to the ``ScheduledJob`` table so the admin UI can display last-run times and statuses. """ +import json import logging import os from datetime import datetime, timedelta, timezone @@ -20,7 +34,15 @@ from pathlib import Path from app.celery_app import celery from app.config import settings from app.database import SessionLocal -from app.models import FileProcessingStep, FileRecord, ScheduledJob +from app.models import ( + FileProcessingStep, + FileRecord, + InAppNotification, + ProcessingLog, + ScheduledJob, + SettingsAuditLog, + SharedLink, +) logger = logging.getLogger(__name__) @@ -270,9 +292,7 @@ def cleanup_temp_files(max_age_hours: int = _TEMP_FILE_MAX_AGE_HOURS) -> dict: active_tmp_filenames: set[str] = set() tmp_dir_str = str(tmp_dir.resolve()) active_records = ( - db.query(FileRecord.local_filename) - .filter(FileRecord.local_filename.like(f"{tmp_dir_str}%")) - .all() + db.query(FileRecord.local_filename).filter(FileRecord.local_filename.like(f"{tmp_dir_str}%")).all() ) for row in active_records: if row.local_filename: @@ -308,11 +328,7 @@ def cleanup_temp_files(max_age_hours: int = _TEMP_FILE_MAX_AGE_HOURS) -> dict: logger.warning("[batch] cleanup_temp_files: could not delete %s: %s", entry, exc) errors += 1 - detail = ( - f"Deleted {deleted} stale temp file(s); " - f"skipped {skipped} (too new or protected); " - f"{errors} error(s)." - ) + detail = f"Deleted {deleted} stale temp file(s); skipped {skipped} (too new or protected); {errors} error(s)." status = "failed" if errors and not deleted else "success" logger.info("[batch] cleanup_temp_files: %s", detail) _update_job_status(job_name, status, detail) @@ -323,3 +339,331 @@ def cleanup_temp_files(max_age_hours: int = _TEMP_FILE_MAX_AGE_HOURS) -> dict: logger.error("[batch] cleanup_temp_files failed: %s", exc, exc_info=True) _update_job_status(job_name, "failed", detail) return {"deleted": 0, "skipped": 0, "errors": 1, "error": str(exc)} + + +# --------------------------------------------------------------------------- +# Task: expire stale shared links +# --------------------------------------------------------------------------- + + +@celery.task(name="app.tasks.batch_tasks.expire_shared_links") +def expire_shared_links() -> dict: + """ + Auto-revoke SharedLinks whose ``expires_at`` timestamp has passed. + + The ``_is_link_valid`` helper in the shared-links API already blocks + access at request time, but the database rows remain flagged as + ``is_active=True``. This task sweeps those rows and sets + ``is_active=False`` + ``revoked_at`` so the management UI reflects + the true state and counts are accurate. + + Returns a summary dict with ``revoked`` count. + """ + job_name = "expire-shared-links" + logger.info("[batch] Starting expire_shared_links task") + + try: + now = datetime.now(timezone.utc) + with SessionLocal() as db: + stale = ( + db.query(SharedLink) + .filter( + SharedLink.is_active.is_(True), + SharedLink.expires_at.isnot(None), + SharedLink.expires_at < now, + ) + .all() + ) + for link in stale: + link.is_active = False + link.revoked_at = now + db.commit() + revoked = len(stale) + + detail = f"Revoked {revoked} expired shared link(s)." + logger.info("[batch] expire_shared_links: %s", detail) + _update_job_status(job_name, "success", detail) + return {"revoked": revoked} + + except Exception as exc: + detail = f"Error: {exc}" + logger.error("[batch] expire_shared_links failed: %s", exc, exc_info=True) + _update_job_status(job_name, "failed", detail) + return {"revoked": 0, "error": str(exc)} + + +# --------------------------------------------------------------------------- +# Task: prune old processing logs +# --------------------------------------------------------------------------- + +#: Default retention period for processing logs and audit log rows. +_LOG_RETENTION_DAYS: int = 30 + + +@celery.task(name="app.tasks.batch_tasks.prune_processing_logs") +def prune_processing_logs(retention_days: int = _LOG_RETENTION_DAYS) -> dict: + """ + Delete ``processing_logs`` and ``settings_audit_log`` rows older than + *retention_days* (default 30) to prevent unbounded table growth. + + Rows for the most recent *retention_days* days are kept so that recent + activity is still visible in the logs/audit UI. + + Args: + retention_days: Number of days of history to keep (default 30). + + Returns: + A summary dict with ``processing_logs_deleted`` and + ``audit_log_deleted`` counts. + """ + job_name = "prune-processing-logs" + logger.info("[batch] Starting prune_processing_logs (retention_days=%s)", retention_days) + + cutoff = datetime.now(timezone.utc) - timedelta(days=retention_days) + + try: + with SessionLocal() as db: + pl_deleted = db.query(ProcessingLog).filter(ProcessingLog.timestamp < cutoff).delete() + al_deleted = db.query(SettingsAuditLog).filter(SettingsAuditLog.changed_at < cutoff).delete() + db.commit() + + detail = ( + f"Deleted {pl_deleted} processing log row(s) and " + f"{al_deleted} settings audit log row(s) older than {retention_days} days." + ) + logger.info("[batch] prune_processing_logs: %s", detail) + _update_job_status(job_name, "success", detail) + return {"processing_logs_deleted": pl_deleted, "audit_log_deleted": al_deleted} + + except Exception as exc: + detail = f"Error: {exc}" + logger.error("[batch] prune_processing_logs failed: %s", exc, exc_info=True) + _update_job_status(job_name, "failed", detail) + return {"processing_logs_deleted": 0, "audit_log_deleted": 0, "error": str(exc)} + + +# --------------------------------------------------------------------------- +# Task: prune old in-app notifications +# --------------------------------------------------------------------------- + +#: Default retention period for read notifications. +_NOTIFICATION_RETENTION_DAYS: int = 30 + + +@celery.task(name="app.tasks.batch_tasks.prune_old_notifications") +def prune_old_notifications(retention_days: int = _NOTIFICATION_RETENTION_DAYS) -> dict: + """ + Delete ``in_app_notifications`` rows that are already read and older than + *retention_days* days (default 30) to prevent unbounded table growth. + + Unread notifications are always kept regardless of age so users do not + miss important alerts. + + Args: + retention_days: Number of days of read-notification history to keep + (default 30). + + Returns: + A summary dict with ``deleted`` count. + """ + job_name = "prune-old-notifications" + logger.info("[batch] Starting prune_old_notifications (retention_days=%s)", retention_days) + + cutoff = datetime.now(timezone.utc) - timedelta(days=retention_days) + + try: + with SessionLocal() as db: + deleted = ( + db.query(InAppNotification) + .filter( + InAppNotification.is_read.is_(True), + InAppNotification.created_at < cutoff, + ) + .delete() + ) + db.commit() + + detail = f"Deleted {deleted} old read notification(s) older than {retention_days} days." + logger.info("[batch] prune_old_notifications: %s", detail) + _update_job_status(job_name, "success", detail) + return {"deleted": deleted} + + except Exception as exc: + detail = f"Error: {exc}" + logger.error("[batch] prune_old_notifications failed: %s", exc, exc_info=True) + _update_job_status(job_name, "failed", detail) + return {"deleted": 0, "error": str(exc)} + + +# --------------------------------------------------------------------------- +# Task: backfill missing AI metadata +# --------------------------------------------------------------------------- + +#: Maximum number of files to process per backfill run. +_METADATA_BACKFILL_BATCH_SIZE: int = 50 + + +@celery.task(name="app.tasks.batch_tasks.backfill_missing_metadata") +def backfill_missing_metadata(batch_size: int = _METADATA_BACKFILL_BATCH_SIZE) -> dict: + """ + Re-trigger AI metadata extraction for documents that have OCR text but + no ``ai_metadata``. + + This handles the common case where a document was processed before the AI + metadata extraction step was configured (e.g., before an OpenAI API key + was added), or where the extraction previously failed. + + Only files that are **not** currently in-progress and have non-empty + ``ocr_text`` are selected. A configurable *batch_size* caps the number + of tasks queued per run to avoid overwhelming the AI provider. + + Args: + batch_size: Maximum number of files to queue per run (default 50). + + Returns: + A summary dict with ``queued`` count. + """ + from app.tasks.extract_metadata_with_gpt import extract_metadata_with_gpt # avoid circular import + + job_name = "backfill-missing-metadata" + logger.info("[batch] Starting backfill_missing_metadata (batch_size=%s)", batch_size) + + try: + with SessionLocal() as db: + # Files currently being processed — skip them. + in_progress_file_ids = ( + db.query(FileProcessingStep.file_id) + .filter(FileProcessingStep.status == "in_progress") + .distinct() + .subquery() + ) + + candidates = ( + db.query(FileRecord) + .filter(FileRecord.is_duplicate.is_(False)) + .filter(FileRecord.ocr_text.isnot(None)) + .filter(FileRecord.ocr_text != "") + .filter((FileRecord.ai_metadata.is_(None)) | (FileRecord.ai_metadata == "")) + .filter(~FileRecord.id.in_(db.query(in_progress_file_ids.c.file_id))) + .limit(batch_size) + .all() + ) + + queued = 0 + for record in candidates: + filename = record.local_filename or record.original_filename or f"file_{record.id}" + extract_metadata_with_gpt.delay( + filename, + record.ocr_text, + file_id=record.id, + ) + queued += 1 + + detail = f"Queued {queued} document(s) for AI metadata backfill." + logger.info("[batch] backfill_missing_metadata: %s", detail) + _update_job_status(job_name, "success", detail) + return {"queued": queued} + + except Exception as exc: + detail = f"Error: {exc}" + logger.error("[batch] backfill_missing_metadata failed: %s", exc, exc_info=True) + _update_job_status(job_name, "failed", detail) + return {"queued": 0, "error": str(exc)} + + +# --------------------------------------------------------------------------- +# Task: sync Meilisearch search index +# --------------------------------------------------------------------------- + +#: Maximum documents to index per sync run. +_SEARCH_SYNC_BATCH_SIZE: int = 100 + + +@celery.task(name="app.tasks.batch_tasks.sync_search_index") +def sync_search_index(batch_size: int = _SEARCH_SYNC_BATCH_SIZE) -> dict: + """ + Index documents in Meilisearch that have OCR text or AI metadata but are + not yet present in the search index. + + This is useful after: + - Enabling Meilisearch for the first time on an existing installation. + - Recovering from a Meilisearch index wipe or migration. + - Documents processed before search indexing was added to the pipeline. + + The task queries the Meilisearch index for existing document IDs, then + finds ``FileRecord`` rows that have processable content (``ocr_text`` or + ``ai_metadata``) but are absent from the index, and re-indexes them. + + A configurable *batch_size* caps the number of documents indexed per run. + + Args: + batch_size: Maximum number of documents to index per run (default 100). + + Returns: + A summary dict with ``indexed`` and ``skipped`` counts. + """ + from app.utils.meilisearch_client import get_meilisearch_client, index_document + + job_name = "sync-search-index" + logger.info("[batch] Starting sync_search_index (batch_size=%s)", batch_size) + + client = get_meilisearch_client() + if client is None: + detail = "Meilisearch is not configured; skipping search index sync." + logger.info("[batch] sync_search_index: %s", detail) + _update_job_status(job_name, "success", detail) + return {"indexed": 0, "skipped": 0, "reason": "meilisearch_not_configured"} + + try: + # Fetch the set of file_ids already in the Meilisearch index. + index = client.get_index(settings.meilisearch_index_name) + # Fetch up to 10 000 IDs — sufficient to determine gaps for most installs. + existing_result = index.get_documents({"fields": ["file_id"], "limit": 10000}) + existing_ids: set[int] = {doc["file_id"] for doc in existing_result.results if "file_id" in doc} + except Exception as exc: + detail = f"Error fetching existing Meilisearch IDs: {exc}" + logger.error("[batch] sync_search_index: %s", detail) + _update_job_status(job_name, "failed", detail) + return {"indexed": 0, "skipped": 0, "error": str(exc)} + + try: + with SessionLocal() as db: + # Files with indexable content that are not already in the index. + candidates = ( + db.query(FileRecord) + .filter(FileRecord.is_duplicate.is_(False)) + .filter( + (FileRecord.ocr_text.isnot(None) & (FileRecord.ocr_text != "")) + | (FileRecord.ai_metadata.isnot(None) & (FileRecord.ai_metadata != "")) + ) + .filter(~FileRecord.id.in_(existing_ids) if existing_ids else True) # type: ignore[arg-type] + .limit(batch_size) + .all() + ) + + indexed = 0 + skipped = 0 + for record in candidates: + metadata: dict = {} + if record.ai_metadata: + try: + metadata = json.loads(record.ai_metadata) + except (json.JSONDecodeError, ValueError): + pass + + success = index_document(record, record.ocr_text or "", metadata) + if success: + indexed += 1 + else: + skipped += 1 + + detail = f"Indexed {indexed} document(s) into Meilisearch; {skipped} skipped (indexing error)." + logger.info("[batch] sync_search_index: %s", detail) + _update_job_status(job_name, "success", detail) + return {"indexed": indexed, "skipped": skipped} + + except Exception as exc: + detail = f"Error: {exc}" + logger.error("[batch] sync_search_index failed: %s", exc, exc_info=True) + _update_job_status(job_name, "failed", detail) + return {"indexed": 0, "skipped": 0, "error": str(exc)} diff --git a/app/views/__init__.py b/app/views/__init__.py index 1faa09fd..5c098527 100644 --- a/app/views/__init__.py +++ b/app/views/__init__.py @@ -24,6 +24,7 @@ from app.views.onedrive import router as onedrive_router from app.views.pipelines import router as pipelines_router # Processing pipelines from app.views.plans import router as plans_router # Admin Plan Designer from app.views.queue import router as queue_router +from app.views.scheduled_jobs import router as scheduled_jobs_router # Scheduled batch jobs from app.views.search import router as search_router from app.views.settings import router as settings_router from app.views.share import router as share_router @@ -58,4 +59,5 @@ router.include_router(pipelines_router) # Processing pipelines router.include_router(imap_accounts_router) # Per-user IMAP ingestion accounts router.include_router(integrations_router) # Unified integrations dashboard router.include_router(notifications_router) # User notification dashboard +router.include_router(scheduled_jobs_router) # Admin scheduled batch jobs router.include_router(help_router) # Built-in help / How-To docs diff --git a/docs/ScheduledJobs.md b/docs/ScheduledJobs.md new file mode 100644 index 00000000..b9bfe0d2 --- /dev/null +++ b/docs/ScheduledJobs.md @@ -0,0 +1,226 @@ +# Scheduled Jobs + +DocuElevate includes an admin-managed **Scheduled Jobs** system that runs +recurring maintenance and processing tasks automatically via Celery Beat. + +All jobs are visible and configurable through the admin UI at +**Admin → Scheduled Jobs** (`/admin/scheduled-jobs`). + +--- + +## Overview + +Scheduled jobs replace the need to run manual batch operations. Each job: + +- Runs automatically on its configured cron or interval schedule. +- Can be **enabled** or **disabled** without restarting the worker. +- Can be triggered **immediately** with the *Run Now* button. +- Reports its last-run time, status (`success`, `failed`, `running`), and + a short detail message back to the UI. + +> **Note:** Schedule changes (cron expressions, interval values, enable/disable +> toggles) are persisted to the database immediately. However, Celery Beat +> reads the schedule only at worker startup — so changes take effect after +> **restarting the Celery worker**. +> The *Run Now* button dispatches a job immediately and does **not** require +> a restart. + +--- + +## Built-in Jobs + +### 1. Process New Documents + +| Field | Value | +|---|---| +| **Task** | `app.tasks.batch_tasks.process_new_documents` | +| **Default schedule** | Every hour (cron `0 */1 * * *`) | +| **Purpose** | Scans for documents that have been uploaded but never processed, then queues them through the full pipeline. | + +A document is considered *new* when it has no `FileProcessingStep` rows +matching the core pipeline steps. Only files whose `local_filename` still +exists on disk are queued; others are skipped and counted. + +--- + +### 2. Reprocess Failed Documents + +| Field | Value | +|---|---| +| **Task** | `app.tasks.batch_tasks.reprocess_failed_documents` | +| **Default schedule** | Every 6 hours (cron `30 */6 * * *`) | +| **Purpose** | Finds documents whose last processing step has status `failure` and re-queues them. Skips files that are currently in-progress. | + +--- + +### 3. Clean Up Temporary Files + +| Field | Value | +|---|---| +| **Task** | `app.tasks.batch_tasks.cleanup_temp_files` | +| **Default schedule** | Daily at 03:30 UTC (cron `30 3 * * *`) | +| **Purpose** | Removes stale files from `workdir/tmp`. Only files older than 24 hours that are not referenced by any active processing job are deleted. | + +--- + +### 4. Expire Stale Shared Links + +| Field | Value | +|---|---| +| **Task** | `app.tasks.batch_tasks.expire_shared_links` | +| **Default schedule** | Daily at 01:00 UTC (cron `0 1 * * *`) | +| **Purpose** | Marks shared document links as inactive when their `expires_at` timestamp has passed. | + +Access is already blocked at request time, but this task keeps the management +UI counts and statuses accurate. + +--- + +### 5. Prune Old Processing Logs + +| Field | Value | +|---|---| +| **Task** | `app.tasks.batch_tasks.prune_processing_logs` | +| **Default schedule** | Weekly on Sunday at 04:00 UTC (cron `0 4 * * 0`) | +| **Purpose** | Deletes `processing_logs` and `settings_audit_log` rows older than 30 days to prevent unbounded table growth. | + +--- + +### 6. Prune Old Notifications + +| Field | Value | +|---|---| +| **Task** | `app.tasks.batch_tasks.prune_old_notifications` | +| **Default schedule** | Weekly on Sunday at 04:30 UTC (cron `30 4 * * 0`) | +| **Purpose** | Deletes *read* `in_app_notifications` rows older than 30 days. Unread notifications are never deleted. | + +--- + +### 7. Backfill Missing AI Metadata + +| Field | Value | +|---|---| +| **Task** | `app.tasks.batch_tasks.backfill_missing_metadata` | +| **Default schedule** | Every 6 hours (cron `0 */6 * * *`) | +| **Purpose** | Re-triggers AI metadata extraction for documents that have OCR text but no `ai_metadata`. Processes up to 50 documents per run. | + +This is particularly useful when: +- An AI provider (OpenAI / Azure) was configured after documents were already + ingested. +- A previous metadata extraction attempt failed. + +--- + +### 8. Sync Search Index + +| Field | Value | +|---|---| +| **Task** | `app.tasks.batch_tasks.sync_search_index` | +| **Default schedule** | Hourly (cron `15 */1 * * *`) | +| **Purpose** | Indexes documents in Meilisearch that have OCR text or AI metadata but are not yet present in the search index. Processes up to 100 documents per run. | + +This is useful after: +- Enabling Meilisearch for the first time on an existing installation. +- Recovering from a Meilisearch index wipe or migration. + +--- + +## Managing Schedules + +### Editing a schedule + +1. Navigate to **Admin → Scheduled Jobs**. +2. Click **Edit Schedule** on the job card. +3. Choose **Cron** or **Interval** and fill in the fields. +4. Click **Save Changes**. + +**Restart the worker** for the new schedule to take effect: + +```bash +# Docker Compose +docker compose restart worker + +# Kubernetes +kubectl rollout restart deployment/docuelevate-worker +``` + +### Cron format + +Cron expressions follow the standard 5-field format: + +``` +minute hour day-of-month month day-of-week +``` + +Examples: + +| Expression | Meaning | +|---|---| +| `0 * * * *` | Every hour on the hour | +| `0 2 * * *` | Daily at 02:00 UTC | +| `30 4 * * 0` | Every Sunday at 04:30 UTC | +| `*/15 * * * *` | Every 15 minutes | + +### Interval format + +For interval schedules, provide the number of **seconds**: + +| Seconds | Meaning | +|---|---| +| `60` | Every minute | +| `3600` | Every hour | +| `86400` | Every day | + +The minimum interval is **60 seconds**. + +--- + +## Enabling / Disabling a job + +Click the toggle on the job card to enable or disable the job. The change +is saved immediately, but the updated schedule only takes effect after the +worker restarts. + +Disabling a job does **not** cancel an in-flight execution — it only prevents +Celery Beat from scheduling new executions. + +--- + +## Run Now + +Click **Run Now** on any job card to dispatch it immediately, regardless of +its schedule or enabled status. The task is sent to the `default` Celery +queue and will be picked up by the next available worker. + +The UI shows a spinner while the dispatch request is in-flight and displays +the Celery task ID on success. Refresh the page after a few seconds to see +the updated last-run status. + +--- + +## Adding custom jobs + +Custom batch jobs can be added by: + +1. Creating a new Celery task in `app/tasks/` (or any existing tasks module). +2. Adding a row to `DEFAULT_JOBS` in `app/api/scheduled_jobs.py`. +3. Running the app — the new job will be seeded automatically on startup. + +The task function should call `_update_job_status(job_name, status, detail)` +at the end of its execution so the admin UI reflects the outcome. + +--- + +## Architecture notes + +- **Seeding**: Default jobs are seeded into the `scheduled_jobs` database + table when the FastAPI application starts up (via the lifespan handler in + `app/main.py`). Seeding is idempotent — re-running does not create + duplicate rows. +- **Beat integration**: `app/celery_worker.py` calls `_load_db_scheduled_jobs()` + at import time to extend `celery.conf.beat_schedule` with enabled DB jobs. + Static hardcoded entries always take precedence over DB entries with the + same key. +- **Status tracking**: Each task calls `_update_job_status()` in + `app/tasks/batch_tasks.py` to persist `last_run_at`, `last_run_status`, and + `last_run_detail` back to the `scheduled_jobs` table. diff --git a/frontend/templates/base.html b/frontend/templates/base.html index c1fd21dc..e4bb200a 100644 --- a/frontend/templates/base.html +++ b/frontend/templates/base.html @@ -165,6 +165,9 @@ Queue Monitor + + Scheduled Jobs + Backup & Restore @@ -332,6 +335,9 @@ Queue Monitor + + Scheduled Jobs + Backup & Restore diff --git a/tests/conftest.py b/tests/conftest.py index d6e3d3aa..ae6db91e 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -67,6 +67,7 @@ from app.models import ( # noqa: F401, E402 PipelineStep, ProcessingLog, SavedSearch, + ScheduledJob, UserImapAccount, UserIntegration, UserProfile, diff --git a/tests/test_scheduled_jobs.py b/tests/test_scheduled_jobs.py new file mode 100644 index 00000000..ef6d218d --- /dev/null +++ b/tests/test_scheduled_jobs.py @@ -0,0 +1,1040 @@ +""" +Tests for the scheduled batch processing feature. + +Covers: +- app/tasks/batch_tasks.py – all 8 batch Celery tasks +- app/api/scheduled_jobs.py – list, update, run-now API endpoints +- app/views/scheduled_jobs.py – admin view route +""" + +import os +from datetime import datetime, timedelta, timezone +from unittest.mock import MagicMock, patch + +import pytest +from fastapi import HTTPException +from fastapi.testclient import TestClient +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker +from sqlalchemy.pool import StaticPool + +from app.database import Base, get_db +from app.models import ( + FileRecord, + InAppNotification, + ProcessingLog, + ScheduledJob, + SettingsAuditLog, + SharedLink, +) + +# --------------------------------------------------------------------------- +# Shared fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture() +def sj_engine(): + """In-memory SQLite engine for scheduled-jobs tests.""" + engine = create_engine( + "sqlite:///:memory:", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + Base.metadata.create_all(bind=engine) + yield engine + Base.metadata.drop_all(bind=engine) + + +@pytest.fixture() +def sj_session(sj_engine): + """DB session scoped to one test.""" + Session = sessionmaker(bind=sj_engine) + session = Session() + yield session + session.close() + + +@pytest.fixture() +def sj_client(sj_engine): + """TestClient with an in-memory DB and admin override.""" + from app.api.scheduled_jobs import _require_admin + from app.main import app + + def override_db(): + Session = sessionmaker(bind=sj_engine) + session = Session() + try: + yield session + finally: + session.close() + + def override_admin(): + return {"email": "admin@example.com", "is_admin": True} + + app.dependency_overrides[get_db] = override_db + app.dependency_overrides[_require_admin] = override_admin + + with TestClient(app, base_url="http://localhost", raise_server_exceptions=False) as client: + yield client + + app.dependency_overrides.clear() + + +@pytest.fixture() +def sj_client_no_admin(sj_engine): + """TestClient with an in-memory DB and no admin override.""" + from app.main import app + + def override_db(): + Session = sessionmaker(bind=sj_engine) + session = Session() + try: + yield session + finally: + session.close() + + app.dependency_overrides[get_db] = override_db + + with TestClient(app, base_url="http://localhost", raise_server_exceptions=False) as client: + yield client + + app.dependency_overrides.clear() + + +def _make_job(session, name="test-job", enabled=True, schedule_type="cron") -> ScheduledJob: + job = ScheduledJob( + name=name, + display_name="Test Job", + description="A test job", + task_name="app.tasks.batch_tasks.cleanup_temp_files", + enabled=enabled, + schedule_type=schedule_type, + cron_minute="0", + cron_hour="*", + cron_day_of_week="*", + cron_day_of_month="*", + cron_month_of_year="*", + ) + session.add(job) + session.commit() + session.refresh(job) + return job + + +def _make_file_record(session, **kwargs) -> FileRecord: + """Insert a minimal FileRecord for testing.""" + defaults = dict( + filehash="abc123", + original_filename="test.pdf", + local_filename="/tmp/test.pdf", + file_size=1024, + mime_type="application/pdf", + is_duplicate=False, + owner_id=None, + ocr_text=None, + ai_metadata=None, + ) + defaults.update(kwargs) + record = FileRecord(**defaults) + session.add(record) + session.commit() + session.refresh(record) + return record + + +# =========================================================================== +# API tests +# =========================================================================== + + +@pytest.mark.unit +class TestListScheduledJobs: + """Tests for GET /api/admin/scheduled-jobs.""" + + def test_returns_empty_list(self, sj_client): + """Returns empty list when no jobs exist.""" + response = sj_client.get("/api/admin/scheduled-jobs") + assert response.status_code == 200 + assert response.json() == [] + + def test_returns_jobs(self, sj_client, sj_session): + """Returns all jobs ordered by display_name.""" + job = _make_job(sj_session) + response = sj_client.get("/api/admin/scheduled-jobs") + assert response.status_code == 200 + data = response.json() + assert len(data) == 1 + assert data[0]["name"] == job.name + assert data[0]["enabled"] is True + + def test_requires_admin(self, sj_client_no_admin): + """Non-admin request receives 403.""" + response = sj_client_no_admin.get("/api/admin/scheduled-jobs") + assert response.status_code == 403 + + +@pytest.mark.unit +class TestUpdateScheduledJob: + """Tests for PATCH /api/admin/scheduled-jobs/{id}.""" + + def test_enable_disable_job(self, sj_client, sj_session): + """PATCH can toggle the enabled flag.""" + job = _make_job(sj_session, enabled=True) + response = sj_client.patch(f"/api/admin/scheduled-jobs/{job.id}", json={"enabled": False}) + assert response.status_code == 200 + assert response.json()["enabled"] is False + + def test_update_cron_schedule(self, sj_client, sj_session): + """PATCH can update cron fields.""" + job = _make_job(sj_session) + payload = { + "schedule_type": "cron", + "cron_minute": "30", + "cron_hour": "6", + "cron_day_of_week": "*", + "cron_day_of_month": "*", + "cron_month_of_year": "*", + } + response = sj_client.patch(f"/api/admin/scheduled-jobs/{job.id}", json=payload) + assert response.status_code == 200 + body = response.json() + assert body["cron_minute"] == "30" + assert body["cron_hour"] == "6" + + def test_update_interval_schedule(self, sj_client, sj_session): + """PATCH can switch to interval schedule.""" + job = _make_job(sj_session) + response = sj_client.patch( + f"/api/admin/scheduled-jobs/{job.id}", + json={"schedule_type": "interval", "interval_seconds": 3600}, + ) + assert response.status_code == 200 + body = response.json() + assert body["schedule_type"] == "interval" + assert body["interval_seconds"] == 3600 + + def test_returns_404_for_missing_job(self, sj_client): + """Returns 404 when the job ID does not exist.""" + response = sj_client.patch("/api/admin/scheduled-jobs/9999", json={"enabled": False}) + assert response.status_code == 404 + + def test_returns_400_for_empty_payload(self, sj_client, sj_session): + """Returns 400 when no updatable fields are provided.""" + job = _make_job(sj_session) + response = sj_client.patch(f"/api/admin/scheduled-jobs/{job.id}", json={}) + assert response.status_code == 400 + + def test_rejects_invalid_schedule_type(self, sj_client, sj_session): + """Returns 422 when schedule_type is not 'cron' or 'interval'.""" + job = _make_job(sj_session) + response = sj_client.patch(f"/api/admin/scheduled-jobs/{job.id}", json={"schedule_type": "invalid"}) + assert response.status_code == 422 + + +@pytest.mark.unit +class TestRunScheduledJobNow: + """Tests for POST /api/admin/scheduled-jobs/{id}/run-now.""" + + def test_dispatches_task(self, sj_client, sj_session): + """run-now sends the task and returns a task_id.""" + job = _make_job(sj_session) + mock_async_result = MagicMock() + mock_async_result.id = "fake-task-id-123" + + with patch("app.celery_app.celery.send_task", return_value=mock_async_result): + response = sj_client.post(f"/api/admin/scheduled-jobs/{job.id}/run-now") + + assert response.status_code == 200 + body = response.json() + assert body["status"] == "dispatched" + assert body["task_id"] == "fake-task-id-123" + assert body["job_name"] == job.name + + def test_returns_404_for_missing_job(self, sj_client): + """Returns 404 when job ID does not exist.""" + response = sj_client.post("/api/admin/scheduled-jobs/9999/run-now") + assert response.status_code == 404 + + +# =========================================================================== +# View tests +# =========================================================================== + + +@pytest.mark.unit +class TestScheduledJobsView: + """Tests for app/views/scheduled_jobs.py.""" + + def test_redirects_non_admin_to_home(self): + """View redirects to '/' when user is not an admin.""" + from app.views.scheduled_jobs import scheduled_jobs_page + + mock_request = MagicMock() + mock_request.session = {"user": {"email": "user@example.com", "is_admin": False}} + + import asyncio + + result = asyncio.get_event_loop().run_until_complete(scheduled_jobs_page(mock_request)) + assert result.status_code == 302 + assert result.headers["location"] == "/" + + def test_redirects_when_no_user_in_session(self): + """View redirects to '/' when no user is in session.""" + from app.views.scheduled_jobs import scheduled_jobs_page + + mock_request = MagicMock() + mock_request.session = {} + + import asyncio + + result = asyncio.get_event_loop().run_until_complete(scheduled_jobs_page(mock_request)) + assert result.status_code == 302 + + def test_returns_template_for_admin(self): + """View returns the scheduled_jobs template for an admin user.""" + from app.views.scheduled_jobs import scheduled_jobs_page + + mock_request = MagicMock() + mock_request.session = {"user": {"email": "admin@example.com", "is_admin": True}} + mock_template_response = MagicMock() + + with patch("app.views.scheduled_jobs.templates") as mock_templates: + mock_templates.TemplateResponse.return_value = mock_template_response + import asyncio + + result = asyncio.get_event_loop().run_until_complete(scheduled_jobs_page(mock_request)) + + mock_templates.TemplateResponse.assert_called_once() + call_args = mock_templates.TemplateResponse.call_args[0] + assert call_args[0] == "admin_scheduled_jobs.html" + assert result is mock_template_response + + def test_raises_500_on_template_error(self): + """View raises HTTPException 500 when template rendering fails.""" + from app.views.scheduled_jobs import scheduled_jobs_page + + mock_request = MagicMock() + mock_request.session = {"user": {"email": "admin@example.com", "is_admin": True}} + + with patch("app.views.scheduled_jobs.templates") as mock_templates: + mock_templates.TemplateResponse.side_effect = RuntimeError("Template not found") + import asyncio + + with pytest.raises(HTTPException) as exc_info: + asyncio.get_event_loop().run_until_complete(scheduled_jobs_page(mock_request)) + + assert exc_info.value.status_code == 500 + assert "Failed to load scheduled jobs page" in exc_info.value.detail + + +# =========================================================================== +# seed_default_scheduled_jobs tests +# =========================================================================== + + +@pytest.mark.unit +class TestSeedDefaultScheduledJobs: + """Tests for seed_default_scheduled_jobs utility.""" + + def test_seeds_all_default_jobs(self, sj_session): + """All DEFAULT_JOBS entries are created on first call.""" + from app.api.scheduled_jobs import DEFAULT_JOBS, seed_default_scheduled_jobs + + seed_default_scheduled_jobs(sj_session) + count = sj_session.query(ScheduledJob).count() + assert count == len(DEFAULT_JOBS) + + def test_is_idempotent(self, sj_session): + """Calling seed twice does not create duplicate entries.""" + from app.api.scheduled_jobs import DEFAULT_JOBS, seed_default_scheduled_jobs + + seed_default_scheduled_jobs(sj_session) + seed_default_scheduled_jobs(sj_session) + count = sj_session.query(ScheduledJob).count() + assert count == len(DEFAULT_JOBS) + + def test_default_jobs_are_enabled(self, sj_session): + """All seeded jobs are enabled by default.""" + from app.api.scheduled_jobs import seed_default_scheduled_jobs + + seed_default_scheduled_jobs(sj_session) + disabled = sj_session.query(ScheduledJob).filter(ScheduledJob.enabled.is_(False)).count() + assert disabled == 0 + + def test_default_jobs_cover_all_batch_tasks(self, sj_session): + """All 8 batch tasks are represented in the default job list.""" + from app.api.scheduled_jobs import DEFAULT_JOBS + + task_names = {j["task_name"] for j in DEFAULT_JOBS} + expected = { + "app.tasks.batch_tasks.process_new_documents", + "app.tasks.batch_tasks.reprocess_failed_documents", + "app.tasks.batch_tasks.cleanup_temp_files", + "app.tasks.batch_tasks.expire_shared_links", + "app.tasks.batch_tasks.prune_processing_logs", + "app.tasks.batch_tasks.prune_old_notifications", + "app.tasks.batch_tasks.backfill_missing_metadata", + "app.tasks.batch_tasks.sync_search_index", + } + assert expected == task_names + + +# =========================================================================== +# Batch task tests +# =========================================================================== + + +@pytest.mark.unit +class TestProcessNewDocuments: + """Tests for batch_tasks.process_new_documents.""" + + def test_returns_success_with_no_candidates(self, sj_engine): + """Returns success with zero queued when no new files exist.""" + from app.tasks.batch_tasks import process_new_documents + + with ( + patch("app.tasks.batch_tasks.SessionLocal") as mock_sl, + patch("app.tasks.batch_tasks._update_job_status") as mock_update, + ): + real_session = sessionmaker(bind=sj_engine)() + mock_sl.return_value.__enter__ = MagicMock(return_value=real_session) + mock_sl.return_value.__exit__ = MagicMock(return_value=False) + + result = process_new_documents() + real_session.close() + + assert result["queued"] == 0 + mock_update.assert_called_once() + assert mock_update.call_args[0][1] == "success" + + def test_handles_db_exception(self): + """DB exceptions are caught and job status is set to failed.""" + from app.tasks.batch_tasks import process_new_documents + + with ( + patch("app.tasks.batch_tasks.SessionLocal") as mock_sl, + patch("app.tasks.batch_tasks._update_job_status") as mock_update, + ): + mock_sl.return_value.__enter__ = MagicMock(side_effect=RuntimeError("DB error")) + mock_sl.return_value.__exit__ = MagicMock(return_value=False) + + result = process_new_documents() + + assert "error" in result + mock_update.assert_called_once() + assert mock_update.call_args[0][1] == "failed" + + def test_skips_file_with_missing_path(self, sj_engine): + """Files whose local_filename does not exist on disk are counted as skipped.""" + from app.tasks.batch_tasks import process_new_documents + + Session = sessionmaker(bind=sj_engine) + session = Session() + _make_file_record(session, filehash="hash_no_file", local_filename="/nonexistent/path.pdf") + session.close() + + with ( + patch("app.tasks.batch_tasks.SessionLocal") as mock_sl, + patch("app.tasks.batch_tasks._update_job_status") as mock_update, + ): + real_session = sessionmaker(bind=sj_engine)() + mock_sl.return_value.__enter__ = MagicMock(return_value=real_session) + mock_sl.return_value.__exit__ = MagicMock(return_value=False) + + result = process_new_documents() + real_session.close() + + assert result["skipped"] == 1 + assert result["queued"] == 0 + mock_update.assert_called_once() + assert mock_update.call_args[0][1] == "success" + + +@pytest.mark.unit +class TestReprocessFailedDocuments: + """Tests for batch_tasks.reprocess_failed_documents.""" + + def test_returns_success_with_no_failed_files(self, sj_engine): + """Returns success with zero queued when no files have failed steps.""" + from app.tasks.batch_tasks import reprocess_failed_documents + + with ( + patch("app.tasks.batch_tasks.SessionLocal") as mock_sl, + patch("app.tasks.batch_tasks._update_job_status") as mock_update, + ): + real_session = sessionmaker(bind=sj_engine)() + mock_sl.return_value.__enter__ = MagicMock(return_value=real_session) + mock_sl.return_value.__exit__ = MagicMock(return_value=False) + + result = reprocess_failed_documents() + real_session.close() + + assert result["queued"] == 0 + mock_update.assert_called_once() + assert mock_update.call_args[0][1] == "success" + + def test_handles_exception(self): + """DB exception sets status to failed.""" + from app.tasks.batch_tasks import reprocess_failed_documents + + with ( + patch("app.tasks.batch_tasks.SessionLocal") as mock_sl, + patch("app.tasks.batch_tasks._update_job_status") as mock_update, + ): + mock_sl.return_value.__enter__ = MagicMock(side_effect=RuntimeError("fail")) + mock_sl.return_value.__exit__ = MagicMock(return_value=False) + + result = reprocess_failed_documents() + + assert "error" in result + mock_update.assert_called_once() + assert mock_update.call_args[0][1] == "failed" + + +@pytest.mark.unit +class TestCleanupTempFiles: + """Tests for batch_tasks.cleanup_temp_files.""" + + def test_deletes_old_unprotected_file(self, tmp_path): + """Old, unreferenced files in workdir/tmp are deleted.""" + from app.tasks.batch_tasks import cleanup_temp_files + + tmp_dir = tmp_path / "tmp" + tmp_dir.mkdir() + old_file = tmp_dir / "old.pdf" + old_file.write_bytes(b"data") + + # Back-date the modification time by 48 hours. + old_mtime = (datetime.now(timezone.utc) - timedelta(hours=48)).timestamp() + os.utime(old_file, (old_mtime, old_mtime)) + + with ( + patch("app.tasks.batch_tasks.SessionLocal") as mock_sl, + patch("app.tasks.batch_tasks._update_job_status"), + patch("app.tasks.batch_tasks.settings") as mock_settings, + ): + mock_settings.workdir = str(tmp_path) + mock_db = MagicMock() + mock_db.__enter__ = MagicMock(return_value=mock_db) + mock_db.__exit__ = MagicMock(return_value=False) + mock_db.query.return_value.join.return_value.filter.return_value.distinct.return_value.all.return_value = [] + mock_db.query.return_value.filter.return_value.all.return_value = [] + mock_sl.return_value = mock_db + + result = cleanup_temp_files(max_age_hours=24) + + assert result["deleted"] == 1 + assert not old_file.exists() + + def test_skips_new_files(self, tmp_path): + """Files younger than max_age_hours are not deleted.""" + from app.tasks.batch_tasks import cleanup_temp_files + + tmp_dir = tmp_path / "tmp" + tmp_dir.mkdir() + new_file = tmp_dir / "new.pdf" + new_file.write_bytes(b"data") + + with ( + patch("app.tasks.batch_tasks.SessionLocal") as mock_sl, + patch("app.tasks.batch_tasks._update_job_status"), + patch("app.tasks.batch_tasks.settings") as mock_settings, + ): + mock_settings.workdir = str(tmp_path) + mock_db = MagicMock() + mock_db.__enter__ = MagicMock(return_value=mock_db) + mock_db.__exit__ = MagicMock(return_value=False) + mock_db.query.return_value.join.return_value.filter.return_value.distinct.return_value.all.return_value = [] + mock_db.query.return_value.filter.return_value.all.return_value = [] + mock_sl.return_value = mock_db + + result = cleanup_temp_files(max_age_hours=24) + + assert result["skipped"] >= 1 + assert new_file.exists() + + def test_missing_tmp_dir(self, tmp_path): + """Returns success immediately when workdir/tmp does not exist.""" + from app.tasks.batch_tasks import cleanup_temp_files + + with ( + patch("app.tasks.batch_tasks._update_job_status") as mock_update, + patch("app.tasks.batch_tasks.settings") as mock_settings, + ): + mock_settings.workdir = str(tmp_path / "nonexistent") + result = cleanup_temp_files() + + assert result["deleted"] == 0 + mock_update.assert_called_once() + assert mock_update.call_args[0][1] == "success" + + def test_skips_protected_file(self, tmp_path): + """Files referenced by in-progress steps are not deleted.""" + from app.tasks.batch_tasks import cleanup_temp_files + + tmp_dir = tmp_path / "tmp" + tmp_dir.mkdir() + protected = tmp_dir / "protected.pdf" + protected.write_bytes(b"data") + + old_mtime = (datetime.now(timezone.utc) - timedelta(hours=48)).timestamp() + os.utime(protected, (old_mtime, old_mtime)) + + with ( + patch("app.tasks.batch_tasks.SessionLocal") as mock_sl, + patch("app.tasks.batch_tasks._update_job_status"), + patch("app.tasks.batch_tasks.settings") as mock_settings, + ): + mock_settings.workdir = str(tmp_path) + mock_db = MagicMock() + mock_db.__enter__ = MagicMock(return_value=mock_db) + mock_db.__exit__ = MagicMock(return_value=False) + + in_progress_row = MagicMock() + in_progress_row.local_filename = str(protected) + mock_db.query.return_value.join.return_value.filter.return_value.distinct.return_value.all.return_value = [ + in_progress_row + ] + mock_db.query.return_value.filter.return_value.all.return_value = [] + mock_sl.return_value = mock_db + + result = cleanup_temp_files(max_age_hours=24) + + assert protected.exists() + assert result["deleted"] == 0 + + +@pytest.mark.unit +class TestExpireSharedLinks: + """Tests for batch_tasks.expire_shared_links.""" + + def test_revokes_expired_links(self, sj_engine): + """Links whose expires_at is in the past are revoked.""" + from app.tasks.batch_tasks import expire_shared_links + + Session = sessionmaker(bind=sj_engine) + session = Session() + # SharedLink.file_id is NOT NULL — create a file record first. + file_rec = _make_file_record(session, filehash="hash_sl_expire") + link = SharedLink( + token="abc123token", + file_id=file_rec.id, + owner_id="user1", + is_active=True, + expires_at=datetime.now(timezone.utc) - timedelta(hours=1), + ) + session.add(link) + session.commit() + link_id = link.id + session.close() + + with ( + patch("app.tasks.batch_tasks.SessionLocal") as mock_sl, + patch("app.tasks.batch_tasks._update_job_status") as mock_update, + ): + real_session = sessionmaker(bind=sj_engine)() + mock_sl.return_value.__enter__ = MagicMock(return_value=real_session) + mock_sl.return_value.__exit__ = MagicMock(return_value=False) + + result = expire_shared_links() + real_session.close() + + assert result["revoked"] == 1 + mock_update.assert_called_once() + assert mock_update.call_args[0][1] == "success" + + check = sessionmaker(bind=sj_engine)() + updated = check.query(SharedLink).filter(SharedLink.id == link_id).first() + assert updated.is_active is False + assert updated.revoked_at is not None + check.close() + + def test_does_not_touch_active_links(self, sj_engine): + """Links with no expires_at are not affected.""" + from app.tasks.batch_tasks import expire_shared_links + + Session = sessionmaker(bind=sj_engine) + session = Session() + file_rec = _make_file_record(session, filehash="hash_sl_active") + link = SharedLink(token="neverexpires", file_id=file_rec.id, owner_id="u1", is_active=True, expires_at=None) + session.add(link) + session.commit() + session.close() + + with ( + patch("app.tasks.batch_tasks.SessionLocal") as mock_sl, + patch("app.tasks.batch_tasks._update_job_status"), + ): + real_session = sessionmaker(bind=sj_engine)() + mock_sl.return_value.__enter__ = MagicMock(return_value=real_session) + mock_sl.return_value.__exit__ = MagicMock(return_value=False) + + result = expire_shared_links() + real_session.close() + + assert result["revoked"] == 0 + + def test_handles_exception(self): + """DB exception sets status to failed.""" + from app.tasks.batch_tasks import expire_shared_links + + with ( + patch("app.tasks.batch_tasks.SessionLocal") as mock_sl, + patch("app.tasks.batch_tasks._update_job_status") as mock_update, + ): + mock_sl.return_value.__enter__ = MagicMock(side_effect=RuntimeError("fail")) + mock_sl.return_value.__exit__ = MagicMock(return_value=False) + + result = expire_shared_links() + + assert "error" in result + mock_update.assert_called_once() + assert mock_update.call_args[0][1] == "failed" + + +@pytest.mark.unit +class TestPruneProcessingLogs: + """Tests for batch_tasks.prune_processing_logs.""" + + def test_deletes_old_logs(self, sj_engine): + """Old processing_log and audit_log rows are deleted.""" + from app.tasks.batch_tasks import prune_processing_logs + + Session = sessionmaker(bind=sj_engine) + session = Session() + old_ts = datetime.now(timezone.utc) - timedelta(days=40) + for _ in range(3): + session.add(ProcessingLog(file_id=None, task_id="t1", step_name="ocr", status="success", timestamp=old_ts)) + for _ in range(2): + session.add( + SettingsAuditLog( + key="k", old_value="a", new_value="b", changed_by="admin", action="update", changed_at=old_ts + ) + ) + session.commit() + session.close() + + with ( + patch("app.tasks.batch_tasks.SessionLocal") as mock_sl, + patch("app.tasks.batch_tasks._update_job_status") as mock_update, + ): + real_session = sessionmaker(bind=sj_engine)() + mock_sl.return_value.__enter__ = MagicMock(return_value=real_session) + mock_sl.return_value.__exit__ = MagicMock(return_value=False) + result = prune_processing_logs(retention_days=30) + real_session.close() + + assert result["processing_logs_deleted"] == 3 + assert result["audit_log_deleted"] == 2 + mock_update.assert_called_once() + assert mock_update.call_args[0][1] == "success" + + def test_keeps_recent_logs(self, sj_engine): + """Logs within the retention window are not deleted.""" + from app.tasks.batch_tasks import prune_processing_logs + + Session = sessionmaker(bind=sj_engine) + session = Session() + recent_ts = datetime.now(timezone.utc) - timedelta(days=5) + session.add(ProcessingLog(file_id=None, task_id="t2", step_name="ocr", status="success", timestamp=recent_ts)) + session.commit() + session.close() + + with ( + patch("app.tasks.batch_tasks.SessionLocal") as mock_sl, + patch("app.tasks.batch_tasks._update_job_status"), + ): + real_session = sessionmaker(bind=sj_engine)() + mock_sl.return_value.__enter__ = MagicMock(return_value=real_session) + mock_sl.return_value.__exit__ = MagicMock(return_value=False) + result = prune_processing_logs(retention_days=30) + real_session.close() + + assert result["processing_logs_deleted"] == 0 + + def test_handles_exception(self): + """DB exception sets status to failed.""" + from app.tasks.batch_tasks import prune_processing_logs + + with ( + patch("app.tasks.batch_tasks.SessionLocal") as mock_sl, + patch("app.tasks.batch_tasks._update_job_status") as mock_update, + ): + mock_sl.return_value.__enter__ = MagicMock(side_effect=RuntimeError("fail")) + mock_sl.return_value.__exit__ = MagicMock(return_value=False) + result = prune_processing_logs() + + assert "error" in result + mock_update.assert_called_once() + assert mock_update.call_args[0][1] == "failed" + + +@pytest.mark.unit +class TestPruneOldNotifications: + """Tests for batch_tasks.prune_old_notifications.""" + + def test_deletes_old_read_notifications(self, sj_engine): + """Old read notifications are deleted.""" + from app.tasks.batch_tasks import prune_old_notifications + + Session = sessionmaker(bind=sj_engine) + session = Session() + old_ts = datetime.now(timezone.utc) - timedelta(days=40) + for _ in range(4): + session.add( + InAppNotification( + owner_id="u1", event_type="document.processed", title="Done", is_read=True, created_at=old_ts + ) + ) + session.add( + InAppNotification( + owner_id="u1", event_type="document.processed", title="Unread", is_read=False, created_at=old_ts + ) + ) + session.commit() + session.close() + + with ( + patch("app.tasks.batch_tasks.SessionLocal") as mock_sl, + patch("app.tasks.batch_tasks._update_job_status") as mock_update, + ): + real_session = sessionmaker(bind=sj_engine)() + mock_sl.return_value.__enter__ = MagicMock(return_value=real_session) + mock_sl.return_value.__exit__ = MagicMock(return_value=False) + result = prune_old_notifications(retention_days=30) + real_session.close() + + assert result["deleted"] == 4 + mock_update.assert_called_once() + assert mock_update.call_args[0][1] == "success" + + def test_keeps_unread_notifications(self, sj_engine): + """Unread notifications are never deleted.""" + from app.tasks.batch_tasks import prune_old_notifications + + Session = sessionmaker(bind=sj_engine) + session = Session() + old_ts = datetime.now(timezone.utc) - timedelta(days=40) + session.add( + InAppNotification( + owner_id="u1", event_type="document.failed", title="Unread", is_read=False, created_at=old_ts + ) + ) + session.commit() + session.close() + + with ( + patch("app.tasks.batch_tasks.SessionLocal") as mock_sl, + patch("app.tasks.batch_tasks._update_job_status"), + ): + real_session = sessionmaker(bind=sj_engine)() + mock_sl.return_value.__enter__ = MagicMock(return_value=real_session) + mock_sl.return_value.__exit__ = MagicMock(return_value=False) + result = prune_old_notifications(retention_days=30) + real_session.close() + + assert result["deleted"] == 0 + + def test_handles_exception(self): + """DB exception sets status to failed.""" + from app.tasks.batch_tasks import prune_old_notifications + + with ( + patch("app.tasks.batch_tasks.SessionLocal") as mock_sl, + patch("app.tasks.batch_tasks._update_job_status") as mock_update, + ): + mock_sl.return_value.__enter__ = MagicMock(side_effect=RuntimeError("fail")) + mock_sl.return_value.__exit__ = MagicMock(return_value=False) + result = prune_old_notifications() + + assert "error" in result + mock_update.assert_called_once() + assert mock_update.call_args[0][1] == "failed" + + +@pytest.mark.unit +class TestBackfillMissingMetadata: + """Tests for batch_tasks.backfill_missing_metadata.""" + + def test_queues_files_with_missing_metadata(self, sj_engine): + """Files with OCR text but no AI metadata are queued.""" + from app.tasks.batch_tasks import backfill_missing_metadata + + Session = sessionmaker(bind=sj_engine) + session = Session() + _make_file_record(session, filehash="hash_meta", ocr_text="Some extracted text", ai_metadata=None) + session.close() + + with ( + patch("app.tasks.batch_tasks.SessionLocal") as mock_sl, + patch("app.tasks.batch_tasks._update_job_status") as mock_update, + patch("app.tasks.extract_metadata_with_gpt.extract_metadata_with_gpt") as mock_extract, + ): + mock_extract.delay = MagicMock() + real_session = sessionmaker(bind=sj_engine)() + mock_sl.return_value.__enter__ = MagicMock(return_value=real_session) + mock_sl.return_value.__exit__ = MagicMock(return_value=False) + result = backfill_missing_metadata(batch_size=10) + real_session.close() + + assert result["queued"] == 1 + mock_update.assert_called_once() + assert mock_update.call_args[0][1] == "success" + + def test_skips_files_with_existing_metadata(self, sj_engine): + """Files that already have ai_metadata are not queued.""" + from app.tasks.batch_tasks import backfill_missing_metadata + + Session = sessionmaker(bind=sj_engine) + session = Session() + _make_file_record(session, filehash="hash_has_meta", ocr_text="text", ai_metadata='{"document_type":"invoice"}') + session.close() + + with ( + patch("app.tasks.batch_tasks.SessionLocal") as mock_sl, + patch("app.tasks.batch_tasks._update_job_status"), + patch("app.tasks.extract_metadata_with_gpt.extract_metadata_with_gpt") as mock_extract, + ): + mock_extract.delay = MagicMock() + real_session = sessionmaker(bind=sj_engine)() + mock_sl.return_value.__enter__ = MagicMock(return_value=real_session) + mock_sl.return_value.__exit__ = MagicMock(return_value=False) + result = backfill_missing_metadata(batch_size=10) + real_session.close() + + assert result["queued"] == 0 + + def test_handles_exception(self): + """DB exception sets status to failed.""" + from app.tasks.batch_tasks import backfill_missing_metadata + + with ( + patch("app.tasks.batch_tasks.SessionLocal") as mock_sl, + patch("app.tasks.batch_tasks._update_job_status") as mock_update, + ): + mock_sl.return_value.__enter__ = MagicMock(side_effect=RuntimeError("fail")) + mock_sl.return_value.__exit__ = MagicMock(return_value=False) + result = backfill_missing_metadata() + + assert "error" in result + mock_update.assert_called_once() + assert mock_update.call_args[0][1] == "failed" + + +@pytest.mark.unit +class TestSyncSearchIndex: + """Tests for batch_tasks.sync_search_index.""" + + def test_skips_when_meilisearch_not_configured(self): + """Returns success immediately when Meilisearch is not configured.""" + from app.tasks.batch_tasks import sync_search_index + + with ( + patch("app.utils.meilisearch_client.get_meilisearch_client", return_value=None), + patch("app.tasks.batch_tasks._update_job_status") as mock_update, + ): + result = sync_search_index() + + assert result["indexed"] == 0 + assert result.get("reason") == "meilisearch_not_configured" + mock_update.assert_called_once() + assert mock_update.call_args[0][1] == "success" + + def test_indexes_missing_documents(self, sj_engine): + """Documents missing from the index are sent to Meilisearch.""" + from app.tasks.batch_tasks import sync_search_index + + Session = sessionmaker(bind=sj_engine) + session = Session() + _make_file_record( + session, filehash="hash_search", ocr_text="searchable text", ai_metadata='{"document_type":"invoice"}' + ) + session.close() + + mock_client = MagicMock() + mock_index = MagicMock() + mock_client.get_index.return_value = mock_index + get_docs_result = MagicMock() + get_docs_result.results = [] + mock_index.get_documents.return_value = get_docs_result + + with ( + patch("app.utils.meilisearch_client.get_meilisearch_client", return_value=mock_client), + patch("app.utils.meilisearch_client.index_document", return_value=True) as mock_idx, + patch("app.tasks.batch_tasks._update_job_status") as mock_update, + patch("app.tasks.batch_tasks.SessionLocal") as mock_sl, + patch("app.tasks.batch_tasks.settings") as mock_settings, + ): + mock_settings.meilisearch_index_name = "documents" + real_session = sessionmaker(bind=sj_engine)() + mock_sl.return_value.__enter__ = MagicMock(return_value=real_session) + mock_sl.return_value.__exit__ = MagicMock(return_value=False) + result = sync_search_index(batch_size=10) + real_session.close() + + assert result["indexed"] == 1 + mock_idx.assert_called_once() + mock_update.assert_called_once() + assert mock_update.call_args[0][1] == "success" + + def test_handles_meilisearch_fetch_error(self): + """Error fetching existing IDs from Meilisearch results in failed status.""" + from app.tasks.batch_tasks import sync_search_index + + mock_client = MagicMock() + mock_client.get_index.side_effect = RuntimeError("Connection refused") + + with ( + patch("app.utils.meilisearch_client.get_meilisearch_client", return_value=mock_client), + patch("app.tasks.batch_tasks._update_job_status") as mock_update, + patch("app.tasks.batch_tasks.settings") as mock_settings, + ): + mock_settings.meilisearch_index_name = "documents" + result = sync_search_index() + + assert "error" in result + mock_update.assert_called_once() + assert mock_update.call_args[0][1] == "failed" + + +# =========================================================================== +# _update_job_status helper tests +# =========================================================================== + + +@pytest.mark.unit +class TestUpdateJobStatus: + """Tests for the _update_job_status helper.""" + + def test_updates_existing_job(self, sj_session): + """_update_job_status sets last_run_at/status/detail on an existing job.""" + from app.tasks.batch_tasks import _update_job_status + + job = _make_job(sj_session, name="test-update-status") + + with patch("app.tasks.batch_tasks.SessionLocal") as mock_sl: + mock_db = MagicMock() + mock_db.__enter__ = MagicMock(return_value=mock_db) + mock_db.__exit__ = MagicMock(return_value=False) + mock_db.query.return_value.filter.return_value.first.return_value = job + mock_sl.return_value = mock_db + + _update_job_status("test-update-status", "success", "Done.") + + assert job.last_run_status == "success" + assert job.last_run_detail == "Done." + + def test_silently_handles_missing_job(self): + """_update_job_status does not raise when the job does not exist.""" + from app.tasks.batch_tasks import _update_job_status + + with patch("app.tasks.batch_tasks.SessionLocal") as mock_sl: + mock_db = MagicMock() + mock_db.__enter__ = MagicMock(return_value=mock_db) + mock_db.__exit__ = MagicMock(return_value=False) + mock_db.query.return_value.filter.return_value.first.return_value = None + mock_sl.return_value = mock_db + + # Should not raise. + _update_job_status("nonexistent-job", "success", "Done.") From ec77c51cb5ed538d346eebe809cc4e8cf580c94c Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Mon, 9 Mar 2026 19:00:13 +0000 Subject: [PATCH 4/4] =?UTF-8?q?fix:=20code=20standards=20audit=20=E2=80=94?= =?UTF-8?q?=20100%=20coverage,=20Pydantic=20v2=20model=5Fconfig,=20asyncio?= =?UTF-8?q?.run(),=20targeted=20edge-case=20tests?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com> --- app/api/scheduled_jobs.py | 3 +- app/tasks/batch_tasks.py | 2 +- tests/test_scheduled_jobs.py | 593 ++++++++++++++++++++++++++++++++++- 3 files changed, 591 insertions(+), 7 deletions(-) diff --git a/app/api/scheduled_jobs.py b/app/api/scheduled_jobs.py index 56f8b62e..46f41e3e 100644 --- a/app/api/scheduled_jobs.py +++ b/app/api/scheduled_jobs.py @@ -68,8 +68,7 @@ class ScheduledJobResponse(BaseModel): created_at: datetime | None updated_at: datetime | None - class Config: - from_attributes = True + model_config = {"from_attributes": True} class ScheduledJobUpdate(BaseModel): diff --git a/app/tasks/batch_tasks.py b/app/tasks/batch_tasks.py index 006b469b..f595dbeb 100644 --- a/app/tasks/batch_tasks.py +++ b/app/tasks/batch_tasks.py @@ -307,7 +307,7 @@ def cleanup_temp_files(max_age_hours: int = _TEMP_FILE_MAX_AGE_HOURS) -> dict: # Check modification time. try: mtime = datetime.fromtimestamp(entry.stat().st_mtime, tz=timezone.utc) - except OSError: + except OSError: # pragma: no cover – only reachable if file vanishes between iterdir() and stat() skipped += 1 continue diff --git a/tests/test_scheduled_jobs.py b/tests/test_scheduled_jobs.py index ef6d218d..c7d28684 100644 --- a/tests/test_scheduled_jobs.py +++ b/tests/test_scheduled_jobs.py @@ -275,7 +275,7 @@ class TestScheduledJobsView: import asyncio - result = asyncio.get_event_loop().run_until_complete(scheduled_jobs_page(mock_request)) + result = asyncio.run(scheduled_jobs_page(mock_request)) assert result.status_code == 302 assert result.headers["location"] == "/" @@ -288,7 +288,7 @@ class TestScheduledJobsView: import asyncio - result = asyncio.get_event_loop().run_until_complete(scheduled_jobs_page(mock_request)) + result = asyncio.run(scheduled_jobs_page(mock_request)) assert result.status_code == 302 def test_returns_template_for_admin(self): @@ -303,7 +303,7 @@ class TestScheduledJobsView: mock_templates.TemplateResponse.return_value = mock_template_response import asyncio - result = asyncio.get_event_loop().run_until_complete(scheduled_jobs_page(mock_request)) + result = asyncio.run(scheduled_jobs_page(mock_request)) mock_templates.TemplateResponse.assert_called_once() call_args = mock_templates.TemplateResponse.call_args[0] @@ -322,7 +322,7 @@ class TestScheduledJobsView: import asyncio with pytest.raises(HTTPException) as exc_info: - asyncio.get_event_loop().run_until_complete(scheduled_jobs_page(mock_request)) + asyncio.run(scheduled_jobs_page(mock_request)) assert exc_info.value.status_code == 500 assert "Failed to load scheduled jobs page" in exc_info.value.detail @@ -1038,3 +1038,588 @@ class TestUpdateJobStatus: # Should not raise. _update_job_status("nonexistent-job", "success", "Done.") + + +# =========================================================================== +# Additional coverage tests +# =========================================================================== + + +@pytest.mark.unit +class TestRequireAdminSuccess: + """Tests for the _require_admin success path.""" + + def test_returns_user_for_admin(self): + """_require_admin returns the user dict when the user is an admin.""" + from app.api.scheduled_jobs import _require_admin + + admin_user = {"email": "admin@example.com", "is_admin": True} + mock_request = MagicMock() + mock_request.session = {"user": admin_user} + + result = _require_admin(mock_request) + assert result == admin_user + + +@pytest.mark.unit +class TestSeedDefaultScheduledJobsErrors: + """Tests for error handling in seed_default_scheduled_jobs.""" + + def test_rolls_back_on_db_error(self, sj_session): + """seed_default_scheduled_jobs rolls back and logs on commit failure.""" + from app.api.scheduled_jobs import seed_default_scheduled_jobs + + with patch.object(sj_session, "commit", side_effect=RuntimeError("DB commit error")): + with patch("app.api.scheduled_jobs.logger") as mock_logger: + seed_default_scheduled_jobs(sj_session) + mock_logger.error.assert_called_once() + + +@pytest.mark.unit +class TestUpdateScheduledJobErrors: + """Tests for update_scheduled_job DB error handling.""" + + def test_returns_500_on_commit_failure(self, sj_session): + """Returns 500 when the DB commit fails during update.""" + from fastapi import HTTPException + + from app.api.scheduled_jobs import ScheduledJobUpdate, update_scheduled_job + + job = _make_job(sj_session, name="job-to-fail") + + mock_request = MagicMock() + mock_request.session = {"user": {"email": "admin@example.com", "is_admin": True}} + + # Patch the session's commit to raise after the job is found. + with patch.object(sj_session, "commit", side_effect=RuntimeError("commit error")): + with pytest.raises(HTTPException) as exc_info: + update_scheduled_job( + job_id=job.id, + payload=ScheduledJobUpdate(enabled=False), + request=mock_request, + db=sj_session, + _admin={"email": "admin@example.com", "is_admin": True}, + ) + + assert exc_info.value.status_code == 500 + assert "Failed to update scheduled job" in exc_info.value.detail + + +@pytest.mark.unit +class TestProcessNewDocumentsQueuing: + """Tests for the file-queuing path of process_new_documents.""" + + def test_queues_file_that_exists(self, sj_engine, tmp_path): + """A file with no processing steps whose path exists is queued.""" + from app.tasks.batch_tasks import process_new_documents + + Session = sessionmaker(bind=sj_engine) + session = Session() + real_file = tmp_path / "real.pdf" + real_file.write_bytes(b"%PDF") + _make_file_record(session, filehash="hash_real_q", local_filename=str(real_file)) + session.close() + + dispatched = [] + + with ( + patch("app.tasks.batch_tasks.SessionLocal") as mock_sl, + patch("app.tasks.batch_tasks._update_job_status"), + ): + real_session = sessionmaker(bind=sj_engine)() + mock_sl.return_value.__enter__ = MagicMock(return_value=real_session) + mock_sl.return_value.__exit__ = MagicMock(return_value=False) + + # Patch process_document.delay inside the function module. + with patch("app.tasks.process_document.process_document") as mock_pd: + mock_pd.delay = MagicMock(side_effect=lambda *a, **kw: dispatched.append((a, kw))) + result = process_new_documents() + + real_session.close() + + # The file has no steps so it should be queued. + assert result["queued"] == 1 + assert result["skipped"] == 0 + + +@pytest.mark.unit +class TestReprocessFailedDocumentsQueuing: + """Tests for the file-queuing path of reprocess_failed_documents.""" + + def test_queues_failed_file_that_exists(self, sj_engine, tmp_path): + """A file with a failed step whose path exists is re-queued.""" + from app.tasks.batch_tasks import reprocess_failed_documents + + Session = sessionmaker(bind=sj_engine) + session = Session() + real_file = tmp_path / "fail.pdf" + real_file.write_bytes(b"%PDF") + record = _make_file_record(session, filehash="hash_fail_q", local_filename=str(real_file)) + + # Add a failed processing step. + from app.models import FileProcessingStep + + step = FileProcessingStep( + file_id=record.id, + step_name="extract_metadata_with_gpt", + status="failure", + ) + session.add(step) + session.commit() + session.close() + + dispatched = [] + + with ( + patch("app.tasks.batch_tasks.SessionLocal") as mock_sl, + patch("app.tasks.batch_tasks._update_job_status"), + ): + real_session = sessionmaker(bind=sj_engine)() + mock_sl.return_value.__enter__ = MagicMock(return_value=real_session) + mock_sl.return_value.__exit__ = MagicMock(return_value=False) + + with patch("app.tasks.process_document.process_document") as mock_pd: + mock_pd.delay = MagicMock(side_effect=lambda *a, **kw: dispatched.append((a, kw))) + result = reprocess_failed_documents() + + real_session.close() + + assert result["queued"] == 1 + assert result["skipped"] == 0 + + +@pytest.mark.unit +class TestCleanupTempFilesEdgeCases: + """Additional edge-case tests for cleanup_temp_files.""" + + def test_increments_errors_on_unlink_failure(self, tmp_path): + """OSError during file deletion increments the errors counter.""" + from app.tasks.batch_tasks import cleanup_temp_files + + tmp_dir = tmp_path / "tmp" + tmp_dir.mkdir() + old_file = tmp_dir / "locked.pdf" + old_file.write_bytes(b"data") + + old_mtime = (datetime.now(timezone.utc) - timedelta(hours=48)).timestamp() + os.utime(old_file, (old_mtime, old_mtime)) + + with ( + patch("app.tasks.batch_tasks.SessionLocal") as mock_sl, + patch("app.tasks.batch_tasks._update_job_status") as mock_update, + patch("app.tasks.batch_tasks.settings") as mock_settings, + patch("pathlib.Path.unlink", side_effect=OSError("Permission denied")), + ): + mock_settings.workdir = str(tmp_path) + mock_db = MagicMock() + mock_db.__enter__ = MagicMock(return_value=mock_db) + mock_db.__exit__ = MagicMock(return_value=False) + mock_db.query.return_value.join.return_value.filter.return_value.distinct.return_value.all.return_value = [] + mock_db.query.return_value.filter.return_value.all.return_value = [] + mock_sl.return_value = mock_db + + result = cleanup_temp_files(max_age_hours=24) + + assert result["errors"] == 1 + assert result["deleted"] == 0 + # When errors > 0 and deleted == 0, status should be "failed". + mock_update.assert_called_once_with( + "cleanup-temp-files", "failed", pytest.approx(mock_update.call_args[0][2], abs=1e9) + ) + + def test_status_success_when_both_deleted_and_errors(self, tmp_path): + """Status is 'success' when at least one file was deleted even if some had errors.""" + from app.tasks.batch_tasks import cleanup_temp_files + + tmp_dir = tmp_path / "tmp" + tmp_dir.mkdir() + + for name in ["good.pdf", "bad.pdf"]: + f = tmp_dir / name + f.write_bytes(b"data") + old_mtime = (datetime.now(timezone.utc) - timedelta(hours=48)).timestamp() + os.utime(f, (old_mtime, old_mtime)) + + # First unlink call succeeds (returns None); second raises OSError. + unlink_calls = {"n": 0} + + def side_effect(*_args, **_kwargs): + unlink_calls["n"] += 1 + if unlink_calls["n"] > 1: + raise OSError("Permission denied") + + with ( + patch("app.tasks.batch_tasks.SessionLocal") as mock_sl, + patch("app.tasks.batch_tasks._update_job_status") as mock_update, + patch("app.tasks.batch_tasks.settings") as mock_settings, + patch("pathlib.Path.unlink", side_effect=side_effect), + ): + mock_settings.workdir = str(tmp_path) + mock_db = MagicMock() + mock_db.__enter__ = MagicMock(return_value=mock_db) + mock_db.__exit__ = MagicMock(return_value=False) + mock_db.query.return_value.join.return_value.filter.return_value.distinct.return_value.all.return_value = [] + mock_db.query.return_value.filter.return_value.all.return_value = [] + mock_sl.return_value = mock_db + + result = cleanup_temp_files(max_age_hours=24) + + # deleted=1, errors=1 → status must be "success" (errors only → "failed"). + assert result["deleted"] == 1 + assert result["errors"] == 1 + assert mock_update.call_args[0][1] == "success" + + def test_skips_non_file_entries(self, tmp_path): + """Subdirectories inside workdir/tmp are skipped (not counted as deleted).""" + from app.tasks.batch_tasks import cleanup_temp_files + + tmp_dir = tmp_path / "tmp" + tmp_dir.mkdir() + subdir = tmp_dir / "subdir" + subdir.mkdir() + + with ( + patch("app.tasks.batch_tasks.SessionLocal") as mock_sl, + patch("app.tasks.batch_tasks._update_job_status"), + patch("app.tasks.batch_tasks.settings") as mock_settings, + ): + mock_settings.workdir = str(tmp_path) + mock_db = MagicMock() + mock_db.__enter__ = MagicMock(return_value=mock_db) + mock_db.__exit__ = MagicMock(return_value=False) + mock_db.query.return_value.join.return_value.filter.return_value.distinct.return_value.all.return_value = [] + mock_db.query.return_value.filter.return_value.all.return_value = [] + mock_sl.return_value = mock_db + + result = cleanup_temp_files(max_age_hours=24) + + assert result["deleted"] == 0 + assert subdir.exists() + + def test_outer_exception_returns_error_dict(self, tmp_path): + """An unexpected exception in cleanup returns an error dict and sets status failed.""" + from app.tasks.batch_tasks import cleanup_temp_files + + with ( + patch("app.tasks.batch_tasks.SessionLocal", side_effect=RuntimeError("session error")), + patch("app.tasks.batch_tasks._update_job_status") as mock_update, + patch("app.tasks.batch_tasks.settings") as mock_settings, + ): + mock_settings.workdir = str(tmp_path) + # Make tmp_dir exist so the early-return doesn't fire. + (tmp_path / "tmp").mkdir() + result = cleanup_temp_files(max_age_hours=24) + + assert "error" in result + mock_update.assert_called_once() + assert mock_update.call_args[0][1] == "failed" + + +@pytest.mark.unit +class TestSyncSearchIndexAdditional: + """Additional coverage tests for sync_search_index.""" + + def test_counts_skipped_on_indexing_failure(self, sj_engine): + """When index_document returns False, the document is counted as skipped.""" + from app.tasks.batch_tasks import sync_search_index + + Session = sessionmaker(bind=sj_engine) + session = Session() + _make_file_record(session, filehash="hash_idx_fail", ocr_text="text", ai_metadata=None) + session.close() + + mock_client = MagicMock() + mock_index = MagicMock() + mock_client.get_index.return_value = mock_index + get_docs_result = MagicMock() + get_docs_result.results = [] + mock_index.get_documents.return_value = get_docs_result + + with ( + patch("app.utils.meilisearch_client.get_meilisearch_client", return_value=mock_client), + patch("app.utils.meilisearch_client.index_document", return_value=False), + patch("app.tasks.batch_tasks._update_job_status") as mock_update, + patch("app.tasks.batch_tasks.SessionLocal") as mock_sl, + patch("app.tasks.batch_tasks.settings") as mock_settings, + ): + mock_settings.meilisearch_index_name = "documents" + real_session = sessionmaker(bind=sj_engine)() + mock_sl.return_value.__enter__ = MagicMock(return_value=real_session) + mock_sl.return_value.__exit__ = MagicMock(return_value=False) + result = sync_search_index(batch_size=10) + real_session.close() + + assert result["skipped"] == 1 + assert result["indexed"] == 0 + mock_update.assert_called_once() + assert mock_update.call_args[0][1] == "success" + + def test_handles_invalid_ai_metadata_json(self, sj_engine): + """Invalid ai_metadata JSON is handled gracefully — metadata defaults to {}.""" + from app.tasks.batch_tasks import sync_search_index + + Session = sessionmaker(bind=sj_engine) + session = Session() + _make_file_record( + session, + filehash="hash_bad_json", + ocr_text="some text", + ai_metadata="{invalid-json}", + ) + session.close() + + mock_client = MagicMock() + mock_index = MagicMock() + mock_client.get_index.return_value = mock_index + get_docs_result = MagicMock() + get_docs_result.results = [] + mock_index.get_documents.return_value = get_docs_result + + with ( + patch("app.utils.meilisearch_client.get_meilisearch_client", return_value=mock_client), + patch("app.utils.meilisearch_client.index_document", return_value=True) as mock_idx, + patch("app.tasks.batch_tasks._update_job_status") as mock_update, + patch("app.tasks.batch_tasks.SessionLocal") as mock_sl, + patch("app.tasks.batch_tasks.settings") as mock_settings, + ): + mock_settings.meilisearch_index_name = "documents" + real_session = sessionmaker(bind=sj_engine)() + mock_sl.return_value.__enter__ = MagicMock(return_value=real_session) + mock_sl.return_value.__exit__ = MagicMock(return_value=False) + result = sync_search_index(batch_size=10) + real_session.close() + + # The record should still be indexed with empty metadata. + assert result["indexed"] == 1 + call_args = mock_idx.call_args + assert call_args[0][2] == {} # metadata is empty dict + mock_update.assert_called_once() + assert mock_update.call_args[0][1] == "success" + + +@pytest.mark.unit +class TestReprocessFailedDocumentsMissingFile: + """Test reprocess_failed_documents when file exists as candidate but not on disk.""" + + def test_skips_failed_file_with_missing_path(self, sj_engine): + """A file with a failed step but no file on disk is counted as skipped.""" + from app.tasks.batch_tasks import reprocess_failed_documents + + Session = sessionmaker(bind=sj_engine) + session = Session() + record = _make_file_record( + session, + filehash="hash_fail_skip", + local_filename="/nonexistent/fail.pdf", + ) + from app.models import FileProcessingStep + + step = FileProcessingStep( + file_id=record.id, + step_name="extract_metadata_with_gpt", + status="failure", + ) + session.add(step) + session.commit() + session.close() + + with ( + patch("app.tasks.batch_tasks.SessionLocal") as mock_sl, + patch("app.tasks.batch_tasks._update_job_status") as mock_update, + ): + real_session = sessionmaker(bind=sj_engine)() + mock_sl.return_value.__enter__ = MagicMock(return_value=real_session) + mock_sl.return_value.__exit__ = MagicMock(return_value=False) + result = reprocess_failed_documents() + real_session.close() + + assert result["skipped"] == 1 + assert result["queued"] == 0 + mock_update.assert_called_once() + assert mock_update.call_args[0][1] == "success" + + +@pytest.mark.unit +class TestCleanupTempFilesProtectedByActivity: + """Tests for cleanup_temp_files protection via in-progress and active tmp records.""" + + def test_protects_file_referenced_by_in_progress_step(self, tmp_path, sj_engine): + """A file with an in-progress step is not deleted.""" + from app.tasks.batch_tasks import cleanup_temp_files + + tmp_dir = tmp_path / "tmp" + tmp_dir.mkdir() + protected = tmp_dir / "inprogress.pdf" + protected.write_bytes(b"data") + old_mtime = (datetime.now(timezone.utc) - timedelta(hours=48)).timestamp() + os.utime(protected, (old_mtime, old_mtime)) + + Session = sessionmaker(bind=sj_engine)() + record = _make_file_record( + Session, + filehash="hash_ip", + local_filename=str(protected), + ) + from app.models import FileProcessingStep + + step = FileProcessingStep( + file_id=record.id, + step_name="extract_text", + status="in_progress", + ) + Session.add(step) + Session.commit() + Session.close() + + with ( + patch("app.tasks.batch_tasks.SessionLocal") as mock_sl, + patch("app.tasks.batch_tasks._update_job_status"), + patch("app.tasks.batch_tasks.settings") as mock_settings, + ): + mock_settings.workdir = str(tmp_path) + real_session = sessionmaker(bind=sj_engine)() + mock_sl.return_value.__enter__ = MagicMock(return_value=real_session) + mock_sl.return_value.__exit__ = MagicMock(return_value=False) + result = cleanup_temp_files(max_age_hours=24) + real_session.close() + + # File should NOT have been deleted. + assert protected.exists() + assert result["deleted"] == 0 + + def test_protects_file_referenced_by_active_tmp_record(self, tmp_path, sj_engine): + """A file listed in FileRecord.local_filename pointing into tmp is not deleted.""" + from app.tasks.batch_tasks import cleanup_temp_files + + tmp_dir = tmp_path / "tmp" + tmp_dir.mkdir() + active_file = tmp_dir / "active.pdf" + active_file.write_bytes(b"data") + old_mtime = (datetime.now(timezone.utc) - timedelta(hours=48)).timestamp() + os.utime(active_file, (old_mtime, old_mtime)) + + Session = sessionmaker(bind=sj_engine)() + _make_file_record( + Session, + filehash="hash_active_tmp", + local_filename=str(active_file), + ) + Session.close() + + with ( + patch("app.tasks.batch_tasks.SessionLocal") as mock_sl, + patch("app.tasks.batch_tasks._update_job_status"), + patch("app.tasks.batch_tasks.settings") as mock_settings, + ): + mock_settings.workdir = str(tmp_path) + real_session = sessionmaker(bind=sj_engine)() + mock_sl.return_value.__enter__ = MagicMock(return_value=real_session) + mock_sl.return_value.__exit__ = MagicMock(return_value=False) + result = cleanup_temp_files(max_age_hours=24) + real_session.close() + + assert active_file.exists() + assert result["deleted"] == 0 + + +@pytest.mark.unit +class TestSyncSearchIndexOuterException: + """Tests for the outer exception handler in sync_search_index.""" + + def test_handles_exception_during_db_query(self): + """If the DB query itself raises, the outer except sets status failed.""" + from app.tasks.batch_tasks import sync_search_index + + mock_client = MagicMock() + mock_index = MagicMock() + mock_client.get_index.return_value = mock_index + get_docs_result = MagicMock() + get_docs_result.results = [] + mock_index.get_documents.return_value = get_docs_result + + with ( + patch("app.utils.meilisearch_client.get_meilisearch_client", return_value=mock_client), + patch("app.tasks.batch_tasks.SessionLocal") as mock_sl, + patch("app.tasks.batch_tasks._update_job_status") as mock_update, + patch("app.tasks.batch_tasks.settings") as mock_settings, + ): + mock_settings.meilisearch_index_name = "documents" + # Make the DB query raise. + mock_sl.return_value.__enter__ = MagicMock(side_effect=RuntimeError("DB down")) + mock_sl.return_value.__exit__ = MagicMock(return_value=False) + + result = sync_search_index() + + assert "error" in result + mock_update.assert_called_once() + assert mock_update.call_args[0][1] == "failed" + + +@pytest.mark.unit +class TestCleanupTempFilesNullLocalFilename: + """Tests for null local_filename rows in cleanup_temp_files inner loops.""" + + def _make_null_row(self): + """Return a mock Row with local_filename=None (as SQLAlchemy returns for NULL cols).""" + row = MagicMock() + row.local_filename = None + return row + + def test_skips_null_local_filename_in_in_progress_records(self, tmp_path): + """In-progress rows with null local_filename do not crash and are ignored.""" + from app.tasks.batch_tasks import cleanup_temp_files + + tmp_dir = tmp_path / "tmp" + tmp_dir.mkdir() + + null_row = self._make_null_row() + + with ( + patch("app.tasks.batch_tasks.SessionLocal") as mock_sl, + patch("app.tasks.batch_tasks._update_job_status"), + patch("app.tasks.batch_tasks.settings") as mock_settings, + ): + mock_settings.workdir = str(tmp_path) + mock_db = MagicMock() + mock_db.__enter__ = MagicMock(return_value=mock_db) + mock_db.__exit__ = MagicMock(return_value=False) + # in_progress_records returns one row with local_filename=None. + mock_db.query.return_value.join.return_value.filter.return_value.distinct.return_value.all.return_value = [ + null_row + ] + # active_records returns empty. + mock_db.query.return_value.filter.return_value.all.return_value = [] + mock_sl.return_value = mock_db + + result = cleanup_temp_files(max_age_hours=24) + + # No crash; the null row is simply ignored. + assert result["deleted"] == 0 + + def test_skips_null_local_filename_in_active_tmp_records(self, tmp_path): + """Active-tmp rows with null local_filename do not crash and are ignored.""" + from app.tasks.batch_tasks import cleanup_temp_files + + tmp_dir = tmp_path / "tmp" + tmp_dir.mkdir() + + null_row = self._make_null_row() + + with ( + patch("app.tasks.batch_tasks.SessionLocal") as mock_sl, + patch("app.tasks.batch_tasks._update_job_status"), + patch("app.tasks.batch_tasks.settings") as mock_settings, + ): + mock_settings.workdir = str(tmp_path) + mock_db = MagicMock() + mock_db.__enter__ = MagicMock(return_value=mock_db) + mock_db.__exit__ = MagicMock(return_value=False) + # in_progress_records returns empty. + mock_db.query.return_value.join.return_value.filter.return_value.distinct.return_value.all.return_value = [] + # active_records returns one row with local_filename=None. + mock_db.query.return_value.filter.return_value.all.return_value = [null_row] + mock_sl.return_value = mock_db + + result = cleanup_temp_files(max_age_hours=24) + + assert result["deleted"] == 0