"""ARQ worker configuration and entrypoint.""" from __future__ import annotations import logging from typing import Any from arq.connections import RedisSettings from arq import cron 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...") # Register search providers (normally done by app startup) try: from app.core.db import get_session_factory from app.plugins.builtins.unified_search.provider_registry import auto_register_providers factory = get_session_factory() async with factory() as db: await auto_register_providers(db) logger.info("Search providers registered for worker") except Exception: logger.warning("Failed to register search providers in worker", exc_info=True) async def on_shutdown(ctx: dict[str, Any]) -> None: """Called when worker shuts down.""" logger.info("ARQ worker shutting down...") # Import job functions directly so ARQ registers them by __name__ from app.plugins.builtins.unified_search.jobs import ( index_mails, index_file, index_contact, index_event, reindex, embedding_batch, ) from app.plugins.builtins.ai_proactive.jobs import deep_analysis from app.plugins.builtins.automation.scheduler import scheduler_tick from app.plugins.builtins.automation.workflow_timeout import check_workflow_timeouts from app.plugins.builtins.automation.agent_runner import run_agent from app.plugins.builtins.automation.execution_engine import run_automation from app.plugins.builtins.tasks.jobs import tasks_due_reminder class WorkerSettings: """ARQ worker settings.""" functions = [ index_mails, index_file, index_contact, index_event, reindex, embedding_batch, deep_analysis, scheduler_tick, check_workflow_timeouts, run_agent, run_automation, tasks_due_reminder, ] redis_settings = _get_redis_settings() on_startup = on_startup on_shutdown = on_shutdown max_jobs = 10 job_timeout = 300 queue_name = "arq:queue" cron_jobs = [ cron(scheduler_tick, second={0, 30}), cron(check_workflow_timeouts, minute={0, 5, 10, 15, 20, 25, 30, 35, 40, 45, 50, 55}), cron(tasks_due_reminder, hour=8, minute=0), ]