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.")