From f1f3ce26b3ee8e307f0da7bc25630cbaed8545de Mon Sep 17 00:00:00 2001
From: Agent Zero <agent@media-on.de>
Date: Sun, 19 Jul 2026 19:36:48 +0200
Subject: [PATCH] feat: add ARQ background worker to container

- Create app/core/worker.py with WorkerSettings for ARQ
- Register all job functions: reindex, index_*, embedding_batch, deep_analysis
- Update prestart.sh to start ARQ worker in background before uvicorn
- Worker runs with max_jobs=10, job_timeout=300s
---
 app/core/worker.py | 88 ++++++++++++++++++++++++++++++++++++++++++++++
 prestart.sh        | 15 ++++----
 2 files changed, 96 insertions(+), 7 deletions(-)
 create mode 100644 app/core/worker.py
 mode change 100644 => 100755 prestart.sh

diff --git a/app/core/worker.py b/app/core/worker.py
new file mode 100644
index 0000000..da1560a
--- /dev/null
+++ b/app/core/worker.py
@@ -0,0 +1,88 @@
+"""ARQ worker configuration and entrypoint."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from arq.connections import RedisSettings
+
+from app.config import get_settings
+
+logger = logging.getLogger(__name__)
+
+
+def _get_redis_settings() -> RedisSettings:
+    """Get Redis settings from app config."""
+    settings = get_settings()
+    return RedisSettings.from_dsn(settings.redis_url)
+
+
+async def on_startup(ctx: dict[str, Any]) -> None:
+    """Called when worker starts."""
+    logger.info("ARQ worker starting...")
+
+
+async def on_shutdown(ctx: dict[str, Any]) -> None:
+    """Called when worker shuts down."""
+    logger.info("ARQ worker shutting down...")
+
+
+def _collect_job_functions() -> dict[str, Any]:
+    """Collect all job functions from plugins and core.
+
+    Each plugin's jobs module may define async functions with the
+    signature (ctx: dict, *args) -> None. We import them all and
+    register by name.
+    """
+    functions: dict[str, Any] = {}
+
+    # Unified search jobs
+    try:
+        from app.plugins.builtins.unified_search.jobs import (
+            index_mails,
+            index_file,
+            index_contact,
+            index_company,
+            index_event,
+            reindex,
+            embedding_batch,
+        )
+        functions.update({
+            "index_mails": index_mails,
+            "index_file": index_file,
+            "index_contact": index_contact,
+            "index_company": index_company,
+            "index_event": index_event,
+            "reindex": reindex,
+            "embedding_batch": embedding_batch,
+        })
+    except Exception:
+        logger.warning("Failed to import unified_search jobs", exc_info=True)
+
+    # AI proactive jobs
+    try:
+        from app.plugins.builtins.ai_proactive.jobs import deep_analysis
+        functions["deep_analysis"] = deep_analysis
+    except Exception:
+        logger.warning("Failed to import ai_proactive jobs", exc_info=True)
+
+    # Calendar jobs (calendar_reminder is enqueued but no handler exists yet)
+    # try:
+    #     from app.plugins.builtins.calendar.jobs import calendar_reminder
+    #     functions["calendar_reminder"] = calendar_reminder
+    # except Exception:
+    #     logger.warning("Failed to import calendar jobs", exc_info=True)
+
+    return functions
+
+
+class WorkerSettings:
+    """ARQ worker settings."""
+    functions = _collect_job_functions()
+    redis_settings = _get_redis_settings()
+    on_startup = on_startup
+    on_shutdown = on_shutdown
+    max_jobs = 10
+    job_timeout = 300
+    queue_name = "arq:queue"
diff --git a/prestart.sh b/prestart.sh
old mode 100644
new mode 100755
index 117f446..9d8563e
--- a/prestart.sh
+++ b/prestart.sh
@@ -4,15 +4,13 @@
 #
 # Responsibilities:
 #   1. Run Alembic DB migrations (alembic upgrade head).
-#   2. Start uvicorn as PID 1 (so signals like SIGTERM are forwarded correctly).
+#   2. Start ARQ background worker in background.
+#   3. Start uvicorn as PID 1 (so signals like SIGTERM are forwarded correctly).
 #
 # Notes:
-#   - `set -e` ensures the container crashes loudly if migrations fail,
-#     rather than starting a broken app on an inconsistent schema.
-#   - `exec uvicorn ...` replaces the shell process with uvicorn, so uvicorn
-#     becomes PID 1 and receives Docker's SIGTERM directly for graceful shutdown.
-#   - `--workers 1` is intentional for v1: SQLite (dev) is single-threaded,
-#     and asyncpg (prod) scales fine with one worker + an async connection pool.
+#   - `set -e` ensures the container crashes loudly if migrations fail.
+#   - ARQ worker runs in background, uvicorn becomes PID 1.
+#   - `--workers 1` is intentional for v1.
 # =============================================================================
 
 set -e
@@ -21,6 +19,9 @@ echo "[prestart] $(date -u +%Y-%m-%dT%H:%M:%SZ) - Running alembic upgrade head..
 alembic upgrade head
 echo "[prestart] DB migrations completed successfully."
 
+echo "[prestart] Starting ARQ background worker..."
+arq app.core.worker.WorkerSettings &
+
 echo "[prestart] Starting uvicorn on 0.0.0.0:8000 (workers=1)..."
 exec uvicorn app.main:app \
     --host 0.0.0.0 \