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>
2026-03-09 18:25:26 +00:00
parent c2d56d7475
commit 5bc1fac79c
10 changed files with 1812 additions and 12 deletions
@@ -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)
@@ -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,
+    },
 ]


@@ -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()
@@ -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

@@ -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)}
@@ -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
@@ -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.
@@ -165,6 +165,9 @@
                  <a href="/admin/queue" role="menuitem" class="flex items-center px-4 py-2 text-sm text-gray-700 hover:bg-gray-100">
                    <i class="fas fa-stream w-4 mr-2 text-blue-500" aria-hidden="true"></i> Queue Monitor
                  </a>
+                  <a href="/admin/scheduled-jobs" role="menuitem" class="flex items-center px-4 py-2 text-sm text-gray-700 hover:bg-gray-100">
+                    <i class="fas fa-clock w-4 mr-2 text-indigo-500" aria-hidden="true"></i> Scheduled Jobs
+                  </a>
                  <a href="/admin/backup" role="menuitem" class="flex items-center px-4 py-2 text-sm text-gray-700 hover:bg-gray-100">
                    <i class="fas fa-database w-4 mr-2 text-green-600" aria-hidden="true"></i> Backup &amp; Restore
                  </a>
@@ -332,6 +335,9 @@
                  <a href="/admin/queue" class="block px-3 py-3 rounded-md text-base font-medium text-gray-700 hover:text-gray-900 hover:bg-gray-50">
                    <i class="fas fa-stream mr-2 text-blue-400" aria-hidden="true"></i> Queue Monitor
                  </a>
+                  <a href="/admin/scheduled-jobs" class="block px-3 py-3 rounded-md text-base font-medium text-gray-700 hover:text-gray-900 hover:bg-gray-50">
+                    <i class="fas fa-clock mr-2 text-indigo-500" aria-hidden="true"></i> Scheduled Jobs
+                  </a>
                  <a href="/admin/backup" class="block px-3 py-3 rounded-md text-base font-medium text-gray-700 hover:text-gray-900 hover:bg-gray-50">
                    <i class="fas fa-database mr-2 text-green-500" aria-hidden="true"></i> Backup &amp; Restore
                  </a>
@@ -67,6 +67,7 @@ from app.models import (  # noqa: F401, E402
    PipelineStep,
    ProcessingLog,
    SavedSearch,
+    ScheduledJob,
    UserImapAccount,
    UserIntegration,
    UserProfile,