"""ARQ worker configuration and entrypoint.""" from __future__ import annotations import logging import traceback from typing import Any from arq import cron from arq.connections import RedisSettings from app.config import get_settings from app.core.job_registry import get_all_jobs, get_job, register_job logger = logging.getLogger(__name__) # ── Distributed lock helpers ───────────────────────────────────────────────── # When multiple worker replicas run concurrently, cron jobs must not fire # on every replica. We use a short-lived Redis SET NX lock per cron call # so only one replica actually executes the job. import uuid # noqa: E402 async def _acquire_cron_lock(job_name: str, ttl_seconds: int = 120) -> str | None: """Try to acquire a distributed lock for a cron job. Returns a lock token (random UUID) if acquired, or None if another replica already holds the lock. The lock auto-expires after *ttl_seconds* to avoid deadlocks if a worker crashes mid-job. """ from app.core.auth import get_redis client = get_redis() token = str(uuid.uuid4()) lock_key = f"leocrm:cron_lock:{job_name}" acquired = await client.set(lock_key, token, nx=True, ex=ttl_seconds) return token if acquired else None async def _release_cron_lock(job_name: str, token: str) -> None: """Release a previously acquired cron lock using a safe compare-and-delete.""" from app.core.auth import get_redis client = get_redis() lock_key = f"leocrm:cron_lock:{job_name}" # Lua script ensures we only delete if the token matches (avoid # releasing a lock that was already expired and re-acquired). script = ( b"if redis.call('get', KEYS[1]) == ARGV[1] " b"then return redis.call('del', KEYS[1]) " b"else return 0 end" ) await client.eval(script, 1, lock_key, token.encode()) def _wrap_cron_with_lock(job_name: str, func: Any, ttl_seconds: int = 120) -> Any: """Wrap a cron callable so it acquires a distributed lock first. If the lock cannot be acquired (another replica is handling it), the wrapped function is silently skipped. """ import functools @functools.wraps(func) async def _locked_wrapper(ctx: dict[str, Any], *args: Any, **kwargs: Any) -> Any: token = await _acquire_cron_lock(job_name, ttl_seconds=ttl_seconds) if token is None: logger.debug("Cron job '%s' skipped — lock held by another replica", job_name) return None try: return await func(ctx, *args, **kwargs) finally: await _release_cron_lock(job_name, token) return _locked_wrapper 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...") # Initialize Redis singleton (same as API lifespan) from app.core.auth import init_redis await init_redis() # Initialize service container from app.core.service_container import get_container container = get_container() await container.initialize() # Initialize plugin registry and discover built-in plugins from sqlalchemy import select as sa_select from app.core.event_bus import get_event_bus from app.core.webhook_dispatcher import register_webhook_event_handlers from app.models.plugin import Plugin as PluginModel from app.plugins.registry import get_registry registry = get_registry() from app.core.db import get_migration_engine migration_engine = get_migration_engine() registry.initialize(migration_engine, app=None) registry.discover_builtins() event_bus = get_event_bus() from app.core.db import get_worker_session_factory async_session = get_worker_session_factory() # Activate plugins that are marked active in DB (register event handlers) # RLS fail-closed requires tenant context for tenant-table writes. # The worker skips plugin activation — cron jobs and contributions # are registered by the API container's startup. The worker only # needs event handlers and job processing. from app.models.tenant import Tenant as TenantModel async with async_session() as db: # Load all tenant IDs for per-tenant event handler registration tenant_result = await db.execute(sa_select(TenantModel.id)) all_tenant_ids = [row[0] for row in tenant_result] logger.info(f"Worker: loaded {len(all_tenant_ids)} tenants") # Register event handlers only for active plugins (no DB writes, no cron job registration) # Load active plugin names from DB (global + tenant-specific) active_plugin_names: set[str] = set() async with async_session() as db: # Global plugins that are marked active result = await db.execute( sa_select(PluginModel.name).where(PluginModel.active.is_(True)) ) active_plugin_names = {row[0] for row in result} logger.info(f"Worker: {len(active_plugin_names)} active plugins: {active_plugin_names}") for name in registry.resolve_load_order(): plugin = registry.get_plugin(name) if plugin is None: continue # Only register event handlers for active plugins if name not in active_plugin_names: logger.debug(f"Worker: skipping event handlers for inactive plugin {name}") continue try: # Just register event handlers, skip DB-writing on_activate if hasattr(plugin, 'register_event_handlers'): await plugin.register_event_handlers(event_bus) logger.info(f"Worker: registered event handlers for {name}") except Exception as exc: logger.warning(f"Worker: failed to register event handlers for {name}: {exc}") # Register webhook dispatcher on the event bus register_webhook_event_handlers(event_bus) logger.info("Worker: webhook event handlers registered") # Register trigger dispatcher — generic event→automation bridge from app.core.trigger_dispatcher import register_trigger_dispatcher register_trigger_dispatcher(event_bus) logger.info("Worker: trigger dispatcher registered") # Register search providers (normally done by app startup) try: from app.plugins.builtins.contracts import get_contract search_contract = get_contract("unified_search") if search_contract is not None: factory = async_session async with factory() as db: await search_contract.auto_register_providers(db) logger.info("Search providers registered for worker") else: logger.debug("Unified search plugin not available — skipping provider registration") 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. Pauses any running WorkflowRun instances so they can be resumed after restart, then closes Redis. """ logger.info("ARQ worker shutting down...") # Pause running workflow instances so they can be resumed after restart try: from sqlalchemy import select as sa_select from app.core.db import get_worker_session_factory from app.models.workflow import WorkflowInstance session_factory = get_worker_session_factory() async with session_factory() as db: result = await db.execute( sa_select(WorkflowInstance).where( WorkflowInstance.status == "running" ) ) running = result.scalars().all() if running: for wf in running: wf.status = "paused" await db.commit() logger.info(f"Paused {len(running)} running workflow(s) for graceful shutdown") except Exception as exc: logger.warning(f"Failed to pause running workflows during shutdown: {exc}") from app.core.auth import close_redis await close_redis() # --------------------------------------------------------------------------- # Lazy-load plugin jobs via importlib so they register themselves with the # job_registry. This avoids circular imports and keeps the worker decoupled # from plugin internals. # --------------------------------------------------------------------------- def _lazy_register_plugin_jobs() -> None: """Import each plugin job module so its register_job() call fires. Dynamically discovers job modules from all registered plugins via get_job_modules() — no hardcoded plugin list (P0-5 fix). """ from app.plugins.registry import get_registry registry = get_registry() # Ensure builtins are discovered if not registry.list_discovered(): registry.discover_builtins() job_modules: list[str] = ["app.core.jobs", "app.core.backup_job", "app.services.import_export_jobs"] for plugin_name in registry.list_discovered(): plugin = registry.get_plugin(plugin_name) if plugin is None: continue job_modules.extend(plugin.get_job_modules()) for mod_name in job_modules: try: import importlib importlib.import_module(mod_name) logger.debug("Lazy-loaded plugin jobs from %s", mod_name) except Exception: logger.warning("Failed to lazy-load plugin jobs from %s", mod_name, exc_info=True) # Trigger lazy registration at module level so jobs are available when # WorkerSettings.functions is evaluated. _lazy_register_plugin_jobs() # ── Outbox processor job ──────────────────────────────────────────────────── async def process_outbox_job(ctx: dict[str, Any]) -> None: """Poll the transactional outbox and publish pending events. Uses a distributed Redis lock so only one worker replica processes the outbox at a time. Runs every 5 seconds. Processes events per-tenant by setting tenant context for RLS. """ from sqlalchemy import text as sa_text from app.core.db import get_worker_session_factory from app.core.outbox import process_outbox_batch factory = get_worker_session_factory() async with factory() as db: try: # Load all tenant IDs for per-tenant outbox processing tenant_result = await db.execute(sa_text("SELECT id FROM tenants")) tenant_ids = [row[0] for row in tenant_result] count = await process_outbox_batch(db, batch_size=50, tenant_ids=tenant_ids) if count: logger.info("Outbox: published %d events", count) except Exception as exc: logger.error("Outbox processing failed", exc_info=True) await db.rollback() # Report to Forgejo via contract (avoid Core→Plugin direct import) try: from app.plugins.builtins.contracts import get_contract reporter_contract = get_contract("forgejo_error_reporter") if reporter_contract is not None: await reporter_contract.report_error_to_forgejo({ "message": f"[Worker] Outbox processing failed: {exc}", "stack": traceback.format_exc(), "context": {"source": "worker_outbox_job"}, }) except Exception: pass # Register the outbox job so it appears in get_all_jobs() register_job("process_outbox", process_outbox_job) # ── Outbox retention cleanup job ───────────────────────────────────────────── async def cleanup_outbox_job(ctx: dict[str, Any]) -> None: """Delete published outbox events older than 30 days. Runs hourly to prevent the outbox table from growing indefinitely. Iterates per-tenant for RLS compliance. """ from sqlalchemy import text as sa_text from app.core.db import get_worker_session_factory from app.core.outbox import cleanup_published_events factory = get_worker_session_factory() async with factory() as db: try: tenant_result = await db.execute(sa_text("SELECT id FROM tenants")) tenant_ids = [row[0] for row in tenant_result] total_deleted = 0 for tenant_id in tenant_ids: await db.execute( sa_text("SELECT set_config('app.current_tenant_id', :tid, true)"), {"tid": str(tenant_id)}, ) deleted = await cleanup_published_events(db, retention_days=30) total_deleted += deleted await db.commit() if total_deleted: logger.info("Outbox retention: cleaned up %d published events", total_deleted) except Exception: logger.error("Outbox retention cleanup failed", exc_info=True) await db.rollback() register_job("cleanup_outbox", cleanup_outbox_job) # ── Audit log retention cleanup job ───────────────────────────────────────── async def cleanup_audit_log_job(ctx: dict[str, Any]) -> None: """Delete audit log entries older than 365 days. Runs daily to prevent the audit_log table from growing indefinitely. Iterates per-tenant for RLS compliance. """ from datetime import UTC, datetime, timedelta from sqlalchemy import delete as sa_delete from sqlalchemy import text as sa_text from app.core.db import get_worker_session_factory from app.models.audit import AuditLog factory = get_worker_session_factory() async with factory() as db: try: tenant_result = await db.execute(sa_text("SELECT id FROM tenants")) tenant_ids = [row[0] for row in tenant_result] cutoff = datetime.now(UTC) - timedelta(days=365) total_deleted = 0 for tenant_id in tenant_ids: await db.execute( sa_text("SELECT set_config('app.current_tenant_id', :tid, true)"), {"tid": str(tenant_id)}, ) result = await db.execute( sa_delete(AuditLog).where(AuditLog.timestamp < cutoff) ) total_deleted += result.rowcount await db.commit() if total_deleted: logger.info("Audit retention: cleaned up %d old entries", total_deleted) except Exception: logger.error("Audit retention cleanup failed", exc_info=True) await db.rollback() register_job("cleanup_audit_log", cleanup_audit_log_job) # ── Trash cleanup job ─────────────────────────────────────────────────────── async def cleanup_trash_job(ctx: dict[str, Any]) -> None: """Permanently delete soft-deleted records older than 90 days. Runs daily to clean up the trash. Iterates per-tenant for RLS compliance. Default retention: 90 days in trash before permanent deletion. """ from datetime import UTC, datetime, timedelta from sqlalchemy import delete as sa_delete from sqlalchemy import text as sa_text from app.core.db import get_worker_session_factory from app.models.entity_attachment import EntityAttachment factory = get_worker_session_factory() async with factory() as db: try: tenant_result = await db.execute(sa_text("SELECT id FROM tenants")) tenant_ids = [row[0] for row in tenant_result] cutoff = datetime.now(UTC) - timedelta(days=90) total_deleted = 0 for tenant_id in tenant_ids: await db.execute( sa_text("SELECT set_config('app.current_tenant_id', :tid, true)"), {"tid": str(tenant_id)}, ) # Delete soft-deleted entity attachments # (Contacts trash cleanup moved to the contacts plugin: # cleanup_contacts_trash — audit P2, no core->contacts import) result = await db.execute( sa_delete(EntityAttachment).where( EntityAttachment.deleted_at.is_not(None), EntityAttachment.deleted_at < cutoff, ) ) total_deleted += result.rowcount await db.commit() if total_deleted: logger.info("Trash cleanup: permanently deleted %d old records", total_deleted) except Exception: logger.error("Trash cleanup failed", exc_info=True) await db.rollback() register_job("cleanup_trash", cleanup_trash_job) # Note: knowledge retention cleanup ("cleanup_knowledge") lives with the # knowledge plugin (app/plugins/builtins/knowledge/jobs.py) and is discovered # via the plugin job-module mechanism — no core→plugin import. class WorkerSettings: """ARQ worker settings.""" functions = get_all_jobs() redis_settings = _get_redis_settings() on_startup = on_startup on_shutdown = on_shutdown max_jobs = 10 max_tries = 3 job_timeout = 300 queue_name = "arq:queue" cron_jobs = [ cron( _wrap_cron_with_lock("scheduler_tick", get_job("scheduler_tick")), minute={0, 5, 10, 15, 20, 25, 30, 35, 40, 45, 50, 55}, ), cron( _wrap_cron_with_lock("tasks_due_reminder", get_job("tasks_due_reminder")), hour=8, minute=0, ), # Outbox processor — every 5 seconds, guarded by distributed lock cron( _wrap_cron_with_lock("process_outbox", process_outbox_job, ttl_seconds=30), second={0, 5, 10, 15, 20, 25, 30, 35, 40, 45, 50, 55}, ), # Outbox retention cleanup — hourly cron( _wrap_cron_with_lock("cleanup_outbox", cleanup_outbox_job, ttl_seconds=300), minute=0, ), # Audit log retention cleanup — daily at 03:00 cron( _wrap_cron_with_lock("cleanup_audit_log", cleanup_audit_log_job, ttl_seconds=300), hour=3, minute=0, ), # Trash cleanup — daily at 04:00 (90 days retention) cron( _wrap_cron_with_lock("cleanup_trash", cleanup_trash_job, ttl_seconds=300), hour=4, minute=0, ), # Contacts trash cleanup — daily at 04:15, owned by the contacts # plugin (audit P2: no core->contacts import in the worker). cron( _wrap_cron_with_lock("cleanup_contacts_trash", get_job("cleanup_contacts_trash"), ttl_seconds=300), hour=4, minute=15, ), # Knowledge retention cleanup — daily at 05:00 (90 days, keeps approved). # Function comes from the knowledge plugin via the job registry. cron( _wrap_cron_with_lock("cleanup_knowledge", get_job("cleanup_knowledge"), ttl_seconds=300), hour=5, minute=0, ), # Scheduled backup — daily at 02:00 (guarded by distributed lock) cron( _wrap_cron_with_lock("run_backup", get_job("run_backup"), ttl_seconds=600), hour=2, minute=0, ), ]