"""
Redis client singleton + distributed job lock.

The lock pattern uses SETNX (SET NX EX) which is atomic in Redis 2.6+.
This prevents concurrent execution of the same scheduled job type.

Lock key  : job:lock:{job_name}
TTL       : expected_max_duration + 60s safety buffer
"""

from __future__ import annotations

import contextlib
from collections.abc import AsyncGenerator

from redis.asyncio import Redis
from redis.asyncio import from_url as redis_from_url

from app.config import get_settings

settings = get_settings()

_redis: Redis | None = None


async def get_redis() -> Redis:
    """Return the application-wide Redis connection (lazy init)."""
    global _redis
    if _redis is None:
        _redis = redis_from_url(
            settings.redis_url,
            encoding="utf-8",
            decode_responses=True,
        )
    return _redis


async def get_redis_dep() -> AsyncGenerator[Redis, None]:
    """FastAPI dependency that yields the Redis client."""
    yield await get_redis()


async def close_redis() -> None:
    """Called on app shutdown to release the connection pool."""
    global _redis
    if _redis is not None:
        await _redis.aclose()
        _redis = None


@contextlib.asynccontextmanager
async def acquire_job_lock(
    job_name: str,
    ttl_seconds: int,
) -> AsyncGenerator[None, None]:
    """
    Async context manager: acquire a Redis SETNX lock for a cron job.

    Raises RuntimeError (→ HTTP 409) if the lock is already held.
    Always releases the lock in the finally block, even on exception.

    Usage:
        async with acquire_job_lock("collect", ttl_seconds=120):
            await run_collection()
    """
    redis = await get_redis()
    lock_key = f"job:lock:{job_name}"

    # SET NX EX — atomic: only set if key does not exist
    acquired = await redis.set(lock_key, "1", nx=True, ex=ttl_seconds)
    if not acquired:
        raise RuntimeError(f"Job '{job_name}' is already running")
    try:
        yield
    finally:
        # Always release, even if the job raised an exception
        await redis.delete(lock_key)