leocrm/app/core/jobs.py

"""ARQ job queue integration."""

from __future__ import annotations

import logging
from typing import Any

from arq import create_pool
from arq.connections import RedisSettings, ArqRedis

from app.config import get_settings

logger = logging.getLogger(__name__)

# ── Global ARQ pool singleton ────────────────────────────────────────────────
_job_pool: ArqRedis | None = None


async def init_job_pool() -> ArqRedis:
    """Create and store the global ARQ job pool.

    Called once during app lifespan startup so every subsequent enqueue
    reuses the same connection instead of opening a new one per call.
    """
    global _job_pool
    if _job_pool is not None:
        logger.warning("init_job_pool() called but pool already initialized")
        return _job_pool
    settings = get_settings()
    redis_settings = RedisSettings.from_dsn(settings.redis_url)
    _job_pool = await create_pool(redis_settings)
    logger.info("Global ARQ job pool initialized")
    return _job_pool


async def close_job_pool() -> None:
    """Close the global ARQ job pool. Called during app lifespan shutdown."""
    global _job_pool
    if _job_pool is not None:
        await _job_pool.close()
        _job_pool = None
        logger.info("Global ARQ job pool closed")


async def get_job_pool() -> ArqRedis:
    """Return the global ARQ job pool singleton.

    If init_job_pool() has not been called yet (e.g. during testing),
    a new pool is created lazily so callers always get a working connection.
    """
    global _job_pool
    if _job_pool is None:
        settings = get_settings()
        redis_settings = RedisSettings.from_dsn(settings.redis_url)
        _job_pool = await create_pool(redis_settings)
        logger.debug("ARQ job pool created lazily (init_job_pool not called)")
    return _job_pool


async def enqueue_job(job_name: str, *args: Any, **kwargs: Any) -> str | None:
    """Enqueue a background job. Returns job_id or None."""
    pool = await get_job_pool()
    job = await pool.enqueue_job(job_name, *args, **kwargs)
    if job:
        return job.job_id
    return None


async def get_job_status(job_id: str) -> dict[str, Any] | None:
    """Get the status of a background job."""
    pool = await get_job_pool()
    job_info = await pool.job_info(job_id)
    if job_info is None:
        return None
    return {
        "job_id": job_id,
        "status": str(job_info.status),
        "result": job_info.result,
        "enqueue_time": job_info.enqueue_time.isoformat() if job_info.enqueue_time else None,
        "start_time": job_info.start_time.isoformat() if job_info.start_time else None,
        "finish_time": job_info.finish_time.isoformat() if job_info.finish_time else None,
    }