app/services/scheduler_service.py

from __future__ import annotations

import asyncio
import re
import secrets
import uuid
from urllib.parse import urlparse
from datetime import UTC, datetime, timedelta
from enum import Enum
from typing import Any

import httpx
from apscheduler.executors.asyncio import AsyncIOExecutor
from apscheduler.jobstores.memory import MemoryJobStore
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.date import DateTrigger
from apscheduler.triggers.interval import IntervalTrigger
from croniter import croniter

from app.config import get_settings
from app.core.logger import get_logger
from app.services.supabase import SupabaseClient, get_supabase_client
from app.utils.http_utils import SharedAsyncClient

logger = get_logger(__name__)
settings = get_settings()


def _redact_url(url: str) -> str:
    parsed = urlparse(url)
    if parsed.password:
        return url.replace(parsed.password, "****")
    return url

# ---------------------------------------------------------------------------
# Redis lazy import helper
# ---------------------------------------------------------------------------

_redis_imported: bool = False
_redis_asyncio: Any = None


def _get_redis_asyncio():
    global _redis_imported, _redis_asyncio
    if not _redis_imported:
        try:
            import redis.asyncio as ra
            _redis_asyncio = ra
        except ImportError:
            _redis_asyncio = None
        _redis_imported = True
    return _redis_asyncio


def _get_redis_jobstore_cls():
    try:
        from apscheduler.jobstores.redis import RedisJobStore
        return RedisJobStore
    except ImportError:
        return None


# ---------------------------------------------------------------------------
# Enumerations
# ---------------------------------------------------------------------------

class JobStatus(str, Enum):
    ACTIVE = "active"
    PAUSED = "paused"
    COMPLETED = "completed"
    FAILED = "failed"
    DELETED = "deleted"


class TriggerType(str, Enum):
    CRON = "cron"
    INTERVAL = "interval"
    DATE = "date"


class HttpMethod(str, Enum):
    GET = "GET"
    POST = "POST"
    PUT = "PUT"
    PATCH = "PATCH"
    DELETE = "DELETE"


class AuthType(str, Enum):
    NONE = "none"
    BEARER = "bearer"
    API_KEY = "api_key"
    BASIC = "basic"
    CUSTOM = "custom"


class ExecutionStatus(str, Enum):
    SUCCESS = "success"
    FAILURE = "failure"
    TIMEOUT = "timeout"


class TriggerReason(str, Enum):
    SCHEDULED = "scheduled"
    MANUAL = "manual"
    RETRY = "retry"


# ---------------------------------------------------------------------------
# SSRF Protection
# ---------------------------------------------------------------------------

_PRIVATE_NETWORKS = [
    re.compile(r"^10\."),
    re.compile(r"^172\.(1[6-9]|2\d|3[01])\."),
    re.compile(r"^192\.168\."),
    re.compile(r"^127\."),
    re.compile(r"^0\."),
    re.compile(r"^169\.254\."),
    re.compile(r"^100\.(6[4-9]|[7-9]\d|1[01]\d|12[0-7])\."),
    re.compile(r"^fc[0-9a-f]{2}:", re.IGNORECASE),
    re.compile(r"^fe80:", re.IGNORECASE),
]

_BLOCKED_HOSTS: set[str] = {
    "localhost", "127.0.0.1", "0.0.0.0", "::1",
    "169.254.169.254",
    "metadata.google.internal",
}


def _check_ssrf(url: str) -> None:
    if not settings.ssrf_protection:
        return
    from urllib.parse import urlparse
    parsed = urlparse(url)
    host = parsed.hostname or ""
    if host.lower() in _BLOCKED_HOSTS:
        raise ValueError(f"SSRF protection: host '{host}' is blocked")
    for pattern in _PRIVATE_NETWORKS:
        if pattern.match(host):
            raise ValueError(f"SSRF protection: host '{host}' resolves to a private network")


# ---------------------------------------------------------------------------
# Auth Header Builder
# ---------------------------------------------------------------------------

def _build_auth_headers(auth_config: dict[str, Any]) -> dict[str, str]:
    auth_headers: dict[str, str] = {}
    auth_type = auth_config.get("type", "none")
    if auth_type == "bearer":
        auth_headers["Authorization"] = f"Bearer {auth_config.get('token', '')}"
    elif auth_type == "api_key":
        header_name = auth_config.get("api_key_header", "X-API-Key")
        auth_headers[header_name] = auth_config.get("api_key", "")
    elif auth_type == "basic":
        import base64
        username = auth_config.get("username", "")
        password = auth_config.get("password", "")
        encoded = base64.b64encode(f"{username}:{password}".encode()).decode()
        auth_headers["Authorization"] = f"Basic {encoded}"
    elif auth_type == "custom":
        custom = auth_config.get("custom_headers") or {}
        auth_headers.update(custom)
    return auth_headers


# ---------------------------------------------------------------------------
# Scheduler Repository (Supabase-backed)
# ---------------------------------------------------------------------------

class SchedulerRepository:
    def __init__(self, client: SupabaseClient):
        self._client = client

    async def create_job(self, data: dict[str, Any]) -> dict[str, Any]:
        result = await self._client.insert("scheduled_jobs", data)
        return result or {}

    async def get_job_by_id(self, job_id: str) -> dict[str, Any] | None:
        try:
            return await self._client.find_one("scheduled_jobs", "id", job_id)
        except Exception:
            return None

    async def get_job_by_name(self, name: str) -> dict[str, Any] | None:
        try:
            return await self._client.find_one("scheduled_jobs", "name", name)
        except Exception:
            return None

    async def list_jobs(
        self,
        status: str | None = None,
        tags: list[str] | None = None,
        page: int = 1,
        page_size: int = 20,
    ) -> tuple[list[dict[str, Any]], int]:
        offset = (page - 1) * page_size
        if status and tags:
            rows = await self._client.select("scheduled_jobs", eq=("status", status), limit=page_size, offset=offset)
            filtered = [r for r in rows if tags and any(t in (r.get("tags") or []) for t in tags)]
            total = len(filtered)
            return filtered[:page_size], total
        elif status:
            rows = await self._client.select("scheduled_jobs", eq=("status", status), limit=page_size, offset=offset)
            total = len(await self._client.select("scheduled_jobs", eq=("status", status)))
            return rows, total
        else:
            rows = await self._client.select("scheduled_jobs", limit=page_size, offset=offset)
            total = len(await self._client.select("scheduled_jobs"))
            return rows, total

    async def update_job(self, job_id: str, data: dict[str, Any]) -> dict[str, Any] | None:
        data["updated_at"] = datetime.now(UTC).isoformat()
        result = await self._client.update("scheduled_jobs", "id", job_id, data)
        return result[0] if result else None

    async def delete_job(self, job_id: str) -> None:
        await self._client.delete("scheduled_jobs", "id", job_id)

    async def get_active_jobs(self) -> list[dict[str, Any]]:
        rows = await self._client.select("scheduled_jobs", eq=("status", JobStatus.ACTIVE.value))
        paused = await self._client.select("scheduled_jobs", eq=("status", JobStatus.PAUSED.value))
        return rows + paused

    async def create_history(self, data: dict[str, Any]) -> dict[str, Any]:
        result = await self._client.insert("job_execution_history", data)
        return result or {}

    async def get_history(
        self,
        job_id: str,
        page: int = 1,
        page_size: int = 20,
    ) -> tuple[list[dict[str, Any]], int]:
        offset = (page - 1) * page_size
        rows = await self._client.select(
            "job_execution_history", eq=("job_id", job_id),
            order=("started_at", True),
            limit=page_size, offset=offset,
        )
        total = await self._client.count("job_execution_history", "job_id", job_id)
        return rows, total

    async def get_history_all(
        self,
        page: int = 1,
        page_size: int = 50,
        status_filter: str | None = None,
    ) -> tuple[list[dict[str, Any]], int]:
        offset = (page - 1) * page_size
        if status_filter:
            rows = await self._client.select(
                "job_execution_history", eq=("status", status_filter),
                order=("started_at", True),
                limit=page_size, offset=offset,
            )
            total = await self._client.count("job_execution_history", "status", status_filter)
        else:
            rows = await self._client.select(
                "job_execution_history",
                order=("started_at", True),
                limit=page_size, offset=offset,
            )
            total = len(await self._client.select("job_execution_history"))
        return rows, total

    async def purge_old_history(self, days: int) -> int:
        cutoff = (datetime.now(UTC) - timedelta(days=days)).isoformat()
        rows = await self._client.select("job_execution_history")
        deleted = 0
        for row in rows:
            created = row.get("created_at", "")
            if created < cutoff:
                await self._client.delete("job_execution_history", "id", row["id"])
                deleted += 1
        return deleted

    async def count_jobs(self, status: str | None = None) -> int:
        if status:
            return await self._client.count("scheduled_jobs", "status", status)
        rows = await self._client.select("scheduled_jobs")
        return len(rows)

    async def count_executions(self, status: str | None = None) -> int:
        if status:
            return await self._client.count("job_execution_history", "status", status)
        rows = await self._client.select("job_execution_history")
        return len(rows)


# ---------------------------------------------------------------------------
# HTTP Execution Engine
# ---------------------------------------------------------------------------

class HttpExecutionResult:
    __slots__ = (
        "success", "status_code", "response_body",
        "response_size", "duration_ms", "error_message",
        "exception_type", "retry_count",
    )

    def __init__(
        self,
        success: bool,
        status_code: int | None = None,
        response_body: str | None = None,
        response_size: int = 0,
        duration_ms: float = 0.0,
        error_message: str | None = None,
        exception_type: str | None = None,
        retry_count: int = 0,
    ) -> None:
        self.success = success
        self.status_code = status_code
        self.response_body = response_body
        self.response_size = response_size
        self.duration_ms = duration_ms
        self.error_message = error_message
        self.exception_type = exception_type
        self.retry_count = retry_count


class HttpExecutionEngine:
    _MAX_RESPONSE_PREVIEW = 500

    def __init__(self) -> None:
        self._http = SharedAsyncClient(
            timeout=httpx.Timeout(settings.max_http_timeout),
            limits=httpx.Limits(
                max_connections=200,
                max_keepalive_connections=50,
                keepalive_expiry=30,
            ),
        )

    async def _get_client(self) -> httpx.AsyncClient:
        return await self._http.get()

    async def close(self) -> None:
        await self._http.close()
        logger.info("HTTP client closed")

    async def execute(self, job: dict[str, Any]) -> HttpExecutionResult:
        retry_on_status: list[int] = job.get("retry_on_status") or [429, 500, 502, 503, 504]
        max_retries: int = job.get("max_retries", 3)
        retry_delay: float = job.get("retry_delay_seconds", 1.0)
        backoff: float = job.get("retry_backoff", 2.0)
        max_delay: float = job.get("retry_max_delay_seconds", 60.0)
        retry_jitter: bool = job.get("retry_jitter", False)

        last_result: HttpExecutionResult | None = None

        for attempt in range(max_retries + 1):
            if attempt > 0:
                delay = min(retry_delay * (backoff ** (attempt - 1)), max_delay)
                if retry_jitter:
                    delay = delay * (0.5 + secrets.randbelow(1000) / 1000.0)
                logger.info("Job retry wait", extra={
                    "job_id": job["id"], "job_name": job["name"],
                    "attempt": attempt, "delay_seconds": round(delay, 2),
                })
                await asyncio.sleep(delay)

            result = await self._execute_once(job, attempt)

            if result.success:
                result.retry_count = attempt
                return result

            last_result = result

            should_retry = False
            if attempt < max_retries:
                if result.exception_type == "TimeoutException" and job.get("retry_on_timeout", True):
                    should_retry = True
                elif result.exception_type in ("ConnectError", "ReadError", "NetworkError") and job.get("retry_on_failure", True):
                    should_retry = True
                elif result.status_code in retry_on_status:
                    should_retry = True

            if not should_retry:
                break

            logger.warning("Job execution retry", extra={
                "job_id": job["id"], "job_name": job["name"],
                "attempt": attempt + 1, "max_retries": max_retries,
                "status_code": result.status_code, "error": result.error_message,
            })

        if last_result is not None:
            last_result.retry_count = max(0, max_retries)
        return last_result or HttpExecutionResult(success=False, error_message="Unknown error")

    async def _execute_once(self, job: dict[str, Any], attempt: int) -> HttpExecutionResult:
        started = datetime.now(UTC)
        client = await self._get_client()

        headers: dict[str, str] = dict(job.get("headers") or {})
        auth_headers = _build_auth_headers(job.get("auth_config") or {})
        headers.update(auth_headers)
        headers.setdefault("User-Agent", f"{settings.app_name}/{settings.app_version}")

        params: dict[str, str] = dict(job.get("query_params") or {})
        method = (job.get("method") or "GET").upper()
        url = job.get("url", "")

        try:
            _check_ssrf(url)
        except ValueError as exc:
            return HttpExecutionResult(
                success=False, error_message=str(exc), exception_type="SSRFViolation",
            )

        request_kwargs: dict[str, Any] = {
            "method": method,
            "url": url,
            "headers": headers,
            "params": params,
            "timeout": job.get("timeout_seconds", 30.0),
        }

        body = job.get("body")
        if body is not None and method in ("POST", "PUT", "PATCH", "DELETE"):
            body_type = job.get("body_type", "json")
            if body_type == "json":
                request_kwargs["json"] = body
            elif body_type == "form":
                request_kwargs["data"] = body
            elif body_type == "raw":
                request_kwargs["content"] = str(body).encode()

        try:
            response = await client.request(**request_kwargs)
            duration_ms = (datetime.now(UTC) - started).total_seconds() * 1000
            response_body = response.text[:self._MAX_RESPONSE_PREVIEW]
            success = response.is_success

            if not success:
                logger.warning("HTTP non-success status", extra={
                    "job_id": job["id"], "job_name": job["name"],
                    "status_code": response.status_code, "attempt": attempt,
                })

            return HttpExecutionResult(
                success=success,
                status_code=response.status_code,
                response_body=response_body,
                response_size=len(response.content),
                duration_ms=duration_ms,
                error_message=None if success else f"HTTP {response.status_code}",
            )

        except httpx.TimeoutException:
            duration_ms = (datetime.now(UTC) - started).total_seconds() * 1000
            return HttpExecutionResult(
                success=False, duration_ms=duration_ms,
                error_message=f"Request timed out after {job.get('timeout_seconds', 30.0)}s",
                exception_type="TimeoutException",
            )
        except httpx.ConnectError as exc:
            return HttpExecutionResult(
                success=False, error_message=f"Connection error: {exc}",
                exception_type="ConnectError",
            )
        except httpx.ReadError as exc:
            return HttpExecutionResult(
                success=False, error_message=f"Read error: {exc}",
                exception_type="ReadError",
            )
        except Exception as exc:
            logger.error("HTTP unexpected error", extra={
                "job_id": job["id"], "job_name": job["name"],
                "attempt": attempt, "error": str(exc),
            })
            return HttpExecutionResult(
                success=False, error_message=str(exc),
                exception_type=type(exc).__name__,
            )


# ---------------------------------------------------------------------------
# Scheduler Service — Production-Grade with Redis HA
# ---------------------------------------------------------------------------

class SchedulerService:
    def __init__(self) -> None:
        self._scheduler: AsyncIOScheduler | None = None
        self._async_redis: Any = None
        self._use_redis = False
        self._instance_id: str = settings.scheduler_instance_id or str(uuid.uuid4())[:8]
        self._http_engine = HttpExecutionEngine()
        self._running_jobs: dict[str, asyncio.Task[None]] = {}
        self._lock = asyncio.Lock()
        self._maintenance_task: asyncio.Task[None] | None = None
        self._repo: SchedulerRepository | None = None

    # -----------------------------------------------------------------------
    # Supabase repository access
    # -----------------------------------------------------------------------

    @staticmethod
    def _get_repo() -> SchedulerRepository:
        """Return a Supabase-backed repository or raise if Supabase is unavailable."""
        client = get_supabase_client()
        if client is None:
            raise RuntimeError("Supabase not available")
        return SchedulerRepository(client)

    @staticmethod
    def _get_optional_repo() -> SchedulerRepository | None:
        """Return a Supabase-backed repository or None if Supabase is unavailable."""
        client = get_supabase_client()
        if client is None:
            return None
        return SchedulerRepository(client)

    # -----------------------------------------------------------------------
    # Redis connection management
    # -----------------------------------------------------------------------

    def _redis_conn_params(self) -> dict:
        """Connection parameters for redis.Redis()."""
        return {
            "host": settings.redis_host,
            "port": settings.redis_port,
            "password": settings.redis_password or None,
            "db": settings.redis_db,
            "ssl": settings.redis_ssl,
            "socket_timeout": settings.redis_socket_timeout,
            "socket_connect_timeout": settings.redis_socket_connect_timeout,
            "retry_on_timeout": settings.redis_retry_on_timeout,
            "health_check_interval": settings.redis_health_check_interval,
        }

    def _redis_jobstore_params(self) -> dict:
        """Connection parameters for RedisJobStore (does not accept url)."""
        if settings.redis_url:
            parsed = urlparse(settings.redis_url)
            return {
                "host": parsed.hostname or settings.redis_host,
                "port": parsed.port or settings.redis_port,
                "password": parsed.password or settings.redis_password or None,
                "db": int(parsed.path.lstrip("/")) if parsed.path and parsed.path != "/" else settings.redis_db,
                "ssl": parsed.scheme in ("rediss", "redis+ssl"),
                "socket_timeout": settings.redis_socket_timeout,
                "socket_connect_timeout": settings.redis_socket_connect_timeout,
                "retry_on_timeout": settings.redis_retry_on_timeout,
                "health_check_interval": settings.redis_health_check_interval,
            }
        return {
            "host": settings.redis_host,
            "port": settings.redis_port,
            "password": settings.redis_password or None,
            "db": settings.redis_db,
            "ssl": settings.redis_ssl,
            "socket_timeout": settings.redis_socket_timeout,
            "socket_connect_timeout": settings.redis_socket_connect_timeout,
            "retry_on_timeout": settings.redis_retry_on_timeout,
            "health_check_interval": settings.redis_health_check_interval,
        }

    async def _connect_async_redis(self) -> Any:
        ra = _get_redis_asyncio()
        if ra is None:
            return None
        try:
            if settings.redis_url:
                client = ra.from_url(settings.redis_url, decode_responses=True)
            else:
                client = ra.Redis(**self._redis_conn_params(), decode_responses=True)
            await client.ping()
            log_url = (
                _redact_url(settings.redis_url)
                if settings.redis_url
                else f"{settings.redis_host}:{settings.redis_port}"
            )
            logger.info("Connected to Redis (url=%s)", log_url)
            return client
        except Exception as exc:
            logger.warning("Redis connection failed: %s — falling back to in-memory scheduler", exc)
            return None

    def _create_redis_jobstore(self) -> Any:
        cls = _get_redis_jobstore_cls()
        if cls is None:
            raise RuntimeError("apscheduler[jobstores_redis] not installed")
        return cls(**self._redis_jobstore_params())

    # -----------------------------------------------------------------------
    # Distributed execution lock (Redis SETNX)
    # -----------------------------------------------------------------------

    def _lock_key(self, job_id: str) -> str:
        return f"{settings.scheduler_coordinator_prefix}lock:{job_id}"

    async def _acquire_execution_lock(self, job: dict[str, Any]) -> bool:
        if not self._use_redis or self._async_redis is None:
            return True
        key = self._lock_key(job["id"])
        ttl = job.get("timeout_seconds", 30) + 30
        acquired = await self._async_redis.setnx(key, self._instance_id)
        if acquired:
            await self._async_redis.expire(key, int(ttl))
            logger.debug("Acquired execution lock for job %s", job["id"])
            return True
        owner = await self._async_redis.get(key)
        logger.warning(
            "Execution lock held by instance %s for job %s — skipping",
            owner, job["id"],
        )
        return False

    async def _release_execution_lock(self, job_id: str) -> None:
        if not self._use_redis or self._async_redis is None:
            return
        key = self._lock_key(job_id)
        await self._async_redis.delete(key)
        logger.debug("Released execution lock for job %s", job_id)

    # -----------------------------------------------------------------------
    # Redis health
    # -----------------------------------------------------------------------

    async def _check_redis_health(self) -> dict[str, Any]:
        if not self._use_redis or self._async_redis is None:
            return {"connected": False, "mode": "memory"}
        try:
            ping = await self._async_redis.ping()
            info = await self._async_redis.info(section="server")
            return {
                "connected": bool(ping),
                "mode": "redis",
                "redis_version": info.get("redis_version", "unknown"),
                "instance_id": self._instance_id,
            }
        except Exception as exc:
            return {"connected": False, "mode": "redis", "error": str(exc)}

    # -----------------------------------------------------------------------
    # Lifecycle
    # -----------------------------------------------------------------------

    async def start(self) -> None:
        self._async_redis = await self._connect_async_redis()
        self._use_redis = self._async_redis is not None

        jobstores: dict[str, Any] = {}
        if self._use_redis:
            try:
                jobstores["default"] = self._create_redis_jobstore()
                logger.info("Using RedisJobStore for job persistence")
            except Exception as exc:
                logger.warning("Failed to create RedisJobStore: %s — falling back to memory", exc)
                self._use_redis = False
                self._async_redis = None

        if not self._use_redis:
            jobstores["default"] = MemoryJobStore()
            logger.info("Using MemoryJobStore (jobs will not survive restart)")

        executors = {"default": AsyncIOExecutor()}
        job_defaults = {
            "coalesce": True,
            "max_instances": 1,
            "misfire_grace_time": settings.scheduler_misfire_grace_time,
        }

        self._scheduler = AsyncIOScheduler(
            jobstores=jobstores,
            executors=executors,
            job_defaults=job_defaults,
        )
        self._scheduler.start()
        logger.info("Scheduler started (mode=%s, instance=%s)", "redis" if self._use_redis else "memory", self._instance_id)

        try:
            repo = self._get_optional_repo()
            if repo is not None:
                jobs = await repo.get_active_jobs()
                restored = 0
                for job in jobs:
                    try:
                        self._schedule_job(job)
                        restored += 1
                    except Exception as exc:
                        logger.error("Failed to restore job %s: %s", job.get("id"), exc)

                if self._use_redis:
                    aps_jobs = self._scheduler.get_jobs()
                    aps_ids = {j.id for j in aps_jobs}
                    db_ids = {f"job_{j['id']}" for j in jobs}
                    stale = aps_ids - db_ids
                    for sid in stale:
                        try:
                            self._scheduler.remove_job(sid)
                            logger.info("Removed stale job %s from Redis store", sid)
                        except Exception:
                            pass

                logger.info("Restored %d jobs from database", restored)
        except Exception as exc:
            logger.warning("Could not restore jobs from database: %s", exc)

        self._maintenance_task = asyncio.create_task(self._maintenance_loop())

    async def shutdown(self) -> None:
        logger.info("Scheduler shutting down (instance=%s)", self._instance_id)
        if self._maintenance_task:
            self._maintenance_task.cancel()
            try:
                await asyncio.wait_for(asyncio.shield(self._maintenance_task), timeout=5.0)
            except (asyncio.CancelledError, asyncio.TimeoutError):
                pass

        if self._scheduler:
            self._scheduler.shutdown(wait=False)

        async with self._lock:
            tasks = list(self._running_jobs.values())
        for task in tasks:
            task.cancel()
            try:
                await asyncio.wait_for(asyncio.shield(task), timeout=5.0)
            except (asyncio.CancelledError, asyncio.TimeoutError):
                pass

        await self._http_engine.close()

        if self._async_redis:
            try:
                await self._async_redis.aclose()
                logger.info("Redis connection closed")
            except Exception:
                pass

        logger.info("Scheduler shutdown complete")

    async def _maintenance_loop(self) -> None:
        while True:
            try:
                await asyncio.sleep(3600)
                if self._use_redis and self._async_redis:
                    try:
                        await self._async_redis.ping()
                    except Exception:
                        logger.error("Redis ping failed in maintenance loop")
                repo = self._get_optional_repo()
                if repo is not None:
                    purged = await repo.purge_old_history(settings.history_retention_days)
                    if purged:
                        logger.info("Purged %d old history records", purged)
            except asyncio.CancelledError:
                break
            except Exception:
                logger.error("Maintenance loop error", exc_info=True)

    # -----------------------------------------------------------------------
    # APScheduler helpers
    # -----------------------------------------------------------------------

    def _build_trigger(self, job: dict[str, Any]):
        tz = job.get("timezone", "UTC")
        cfg = job.get("trigger_config") or {}
        trigger_type = job.get("trigger_type")

        if trigger_type == TriggerType.CRON.value:
            cron_expr = cfg.get("cron_expression", "* * * * *")
            parts = cron_expr.split()
            return CronTrigger(
                second=parts[0] if len(parts) > 0 else "*",
                minute=parts[1] if len(parts) > 1 else "*",
                hour=parts[2] if len(parts) > 2 else "*",
                day=parts[3] if len(parts) > 3 else "*",
                month=parts[4] if len(parts) > 4 else "*",
                day_of_week=parts[5] if len(parts) > 5 else "*",
                timezone=tz,
                start_date=job.get("start_date"),
                end_date=job.get("end_date"),
                jitter=job.get("jitter_seconds"),
            )
        elif trigger_type == TriggerType.INTERVAL.value:
            return IntervalTrigger(
                weeks=cfg.get("weeks", 0),
                days=cfg.get("days", 0),
                hours=cfg.get("hours", 0),
                minutes=cfg.get("minutes", 0),
                seconds=cfg.get("seconds", 0),
                timezone=tz,
                start_date=job.get("start_date"),
                end_date=job.get("end_date"),
                jitter=job.get("jitter_seconds"),
            )
        elif trigger_type == TriggerType.DATE.value:
            run_date = cfg.get("run_date")
            return DateTrigger(run_date=run_date, timezone=tz)
        raise ValueError(f"Unknown trigger type: {trigger_type}")

    def _schedule_job(self, job: dict[str, Any]) -> None:
        if self._scheduler is None:
            raise RuntimeError("Scheduler not started")
        trigger = self._build_trigger(job)
        aps_id = f"job_{job['id']}"
        self._scheduler.add_job(
            func=self._execute_job_wrapper,
            trigger=trigger,
            args=[job["id"]],
            id=aps_id,
            name=job.get("name", "unknown"),
            coalesce=job.get("coalesce", True),
            max_instances=job.get("max_instances", 1),
            misfire_grace_time=job.get("misfire_grace_time") or settings.scheduler_misfire_grace_time,
            replace_existing=True,
        )
        logger.info("Scheduled job %s (%s)", job["name"], job["id"])

    def _reschedule_job(self, job: dict[str, Any]) -> None:
        aps_id = f"job_{job['id']}"
        try:
            self._scheduler.remove_job(aps_id)
        except Exception:
            pass
        if job.get("status") in (JobStatus.ACTIVE.value, JobStatus.PAUSED.value):
            self._schedule_job(job)
            if job.get("status") == JobStatus.PAUSED.value:
                try:
                    self._scheduler.pause_job(aps_id)
                except Exception:
                    pass

    def _remove_aps_job(self, job_id: str) -> None:
        aps_id = f"job_{job_id}"
        try:
            self._scheduler.remove_job(aps_id)
        except Exception:
            pass

    def pause_aps_job(self, job_id: str) -> None:
        aps_id = f"job_{job_id}"
        try:
            self._scheduler.pause_job(aps_id)
        except Exception as exc:
            raise RuntimeError(f"Failed to pause job: {exc}") from exc

    def resume_aps_job(self, job_id: str) -> None:
        aps_id = f"job_{job_id}"
        try:
            self._scheduler.resume_job(aps_id)
        except Exception as exc:
            raise RuntimeError(f"Failed to resume job: {exc}") from exc

    def get_next_run_time(self, job_id: str) -> datetime | None:
        if self._scheduler is None:
            return None
        aps_id = f"job_{job_id}"
        aps_job = self._scheduler.get_job(aps_id)
        return aps_job.next_run_time if aps_job else None

    def get_scheduler_status(self) -> dict[str, Any]:
        if self._scheduler is None:
            return {"running": False, "pending_jobs": 0, "currently_executing": 0}
        return {
            "running": self._scheduler.running,
            "pending_jobs": len(self._scheduler.get_jobs()),
            "currently_executing": len(self._running_jobs),
        }

    async def get_health(self) -> dict[str, Any]:
        redis_status = await self._check_redis_health()
        scheduler_status = self.get_scheduler_status()
        return {
            "scheduler": scheduler_status,
            "redis": redis_status,
            "instance_id": self._instance_id,
            "mode": "redis" if self._use_redis else "memory",
            "running_job_ids": self.get_running_job_ids(),
        }

    def get_running_job_ids(self) -> list[str]:
        return list(self._running_jobs.keys())

    async def _execute_job_wrapper(self, job_id: str) -> None:
        task = asyncio.create_task(self._execute_job(job_id))
        async with self._lock:
            self._running_jobs[job_id] = task
        try:
            await task
        except asyncio.CancelledError:
            logger.warning("Job task cancelled: %s", job_id)
        except Exception:
            logger.error("Job task error: %s", job_id, exc_info=True)
        finally:
            async with self._lock:
                self._running_jobs.pop(job_id, None)

    async def _execute_job(
        self,
        job_id: str,
        trigger_reason: TriggerReason = TriggerReason.SCHEDULED,
    ) -> None:
        started_at = datetime.now(UTC)
        logger.info("Job execution started", extra={
            "job_id": job_id, "trigger_reason": trigger_reason.value,
        })

        repo = self._get_optional_repo()
        if repo is None:
            logger.error("Supabase not available, cannot execute job %s", job_id)
            return

        job = await repo.get_job_by_id(job_id)
        if not job:
            logger.error("Job not found: %s", job_id)
            return

        if job.get("status") not in (JobStatus.ACTIVE.value,):
            logger.warning("Job %s not active, skipping", job_id)
            return

        # Acquire distributed lock (no-op in memory mode)
        if not await self._acquire_execution_lock(job):
            return

        try:
            # Enforce execution timeout
            timeout = job.get("timeout_seconds", 30.0)
            result = await asyncio.wait_for(
                self._http_engine.execute(job),
                timeout=timeout + 10.0,
            )
        except asyncio.TimeoutError:
            result = HttpExecutionResult(
                success=False,
                error_message=f"Job execution timed out after {timeout + 10.0}s",
                exception_type="ExecutionTimeout",
            )
        except Exception as exc:
            result = HttpExecutionResult(
                success=False,
                error_message=str(exc),
                exception_type=type(exc).__name__,
            )
        finally:
            await self._release_execution_lock(job_id)

        ended_at = datetime.now(UTC)
        duration_ms = (ended_at - started_at).total_seconds() * 1000

        exec_status = ExecutionStatus.SUCCESS if result.success else ExecutionStatus.FAILURE
        if result.exception_type in ("TimeoutException", "ExecutionTimeout"):
            exec_status = ExecutionStatus.TIMEOUT

        history_data = {
            "id": str(uuid.uuid4()),
            "job_id": job["id"],
            "job_name": job.get("name", ""),
            "started_at": started_at.isoformat(),
            "ended_at": ended_at.isoformat(),
            "duration_ms": round(duration_ms, 2),
            "status": exec_status.value,
            "http_status_code": result.status_code,
            "response_size_bytes": result.response_size,
            "retry_count": result.retry_count,
            "trigger_reason": trigger_reason.value,
            "error_message": result.error_message,
            "exception_type": result.exception_type,
            "request_url": job.get("url", ""),
            "request_method": job.get("method", "GET"),
            "response_preview": result.response_body,
        }
        await repo.create_history(history_data)

        update_data: dict[str, Any] = {
            "last_run_at": started_at.isoformat(),
            "run_count": (job.get("run_count") or 0) + 1,
        }
        if result.success:
            update_data["success_count"] = (job.get("success_count") or 0) + 1
        else:
            update_data["failure_count"] = (job.get("failure_count") or 0) + 1

        next_run = self.get_next_run_time(job_id)
        if next_run:
            update_data["next_run_at"] = next_run.isoformat()

        await repo.update_job(job_id, update_data)

        if result.success:
            logger.info("Job execution success", extra={
                "job_id": job_id, "job_name": job.get("name"),
                "duration_ms": round(duration_ms, 2),
                "http_status": result.status_code,
                "response_size": result.response_size,
                "retry_count": result.retry_count,
            })
        else:
            logger.error("Job execution failure", extra={
                "job_id": job_id, "job_name": job.get("name"),
                "duration_ms": round(duration_ms, 2),
                "http_status": result.status_code,
                "error": result.error_message,
                "exception_type": result.exception_type,
                "retry_count": result.retry_count,
            })

    # -----------------------------------------------------------------------
    # Public API: Job CRUD
    # -----------------------------------------------------------------------

    async def create_job(self, body: dict[str, Any]) -> dict[str, Any]:
        repo = self._get_repo()

        existing = await repo.get_job_by_name(body["name"])
        if existing:
            raise ValueError(f"Job with name '{body['name']}' already exists")

        _check_ssrf(body["url"])

        job_id = str(uuid.uuid4())
        now = datetime.now(UTC).isoformat()

        trigger_config = dict(body.get("trigger", {}))
        trigger_type = trigger_config.pop("type", "cron")

        retry = body.get("retry", {})

        job_data: dict[str, Any] = {
            "id": job_id,
            "name": body["name"],
            "description": body.get("description"),
            "url": body["url"],
            "method": body.get("method", "GET"),
            "headers": body.get("headers", {}),
            "query_params": body.get("query_params", {}),
            "body": body.get("body"),
            "body_type": body.get("body_type", "json"),
            "auth_type": body.get("auth", {}).get("type", "none"),
            "auth_config": dict(body.get("auth", {})),
            "trigger_type": trigger_type,
            "trigger_config": trigger_config,
            "timezone": body.get("timezone", "UTC"),
            "timeout_seconds": body.get("timeout", 30.0),
            "max_retries": retry.get("max_retries", 3),
            "retry_delay_seconds": retry.get("retry_delay", 1.0),
            "retry_backoff": retry.get("retry_backoff", 2.0),
            "retry_max_delay_seconds": retry.get("retry_max_delay", 60.0),
            "retry_on_status": retry.get("retry_on_status", [429, 500, 502, 503, 504]),
            "retry_on_timeout": retry.get("retry_on_timeout", True),
            "retry_on_failure": retry.get("retry_on_failure", True),
            "retry_jitter": retry.get("retry_jitter", False),
            "max_instances": body.get("max_instances", 1),
            "coalesce": body.get("coalesce", True),
            "misfire_grace_time": body.get("misfire_grace_time"),
            "jitter_seconds": trigger_config.get("jitter"),
            "start_date": trigger_config.get("start_date"),
            "end_date": trigger_config.get("end_date"),
            "status": JobStatus.ACTIVE.value,
            "tags": body.get("tags", []),
            "metadata": body.get("metadata", {}),
            "created_at": now,
            "updated_at": now,
            "run_count": 0,
            "failure_count": 0,
            "success_count": 0,
        }

        created = await repo.create_job(job_data)
        self._schedule_job(job_data)

        next_run = self.get_next_run_time(job_id)
        if next_run:
            await repo.update_job(job_id, {"next_run_at": next_run.isoformat()})
            created["next_run_at"] = next_run.isoformat()

        logger.info("Job created: %s (%s)", body["name"], job_id)
        return await repo.get_job_by_id(job_id) or created

    async def update_job(self, job_id: str, body: dict[str, Any]) -> dict[str, Any]:
        repo = self._get_repo()
        job = await repo.get_job_by_id(job_id)
        if not job:
            raise KeyError(f"Job '{job_id}' not found")
        if job.get("status") == JobStatus.DELETED.value:
            raise ValueError("Cannot update a deleted job")

        update_data: dict[str, Any] = {}

        field_map = {
            "description": "description",
            "url": "url",
            "method": "method",
            "headers": "headers",
            "query_params": "query_params",
            "body": "body",
            "body_type": "body_type",
            "timezone": "timezone",
            "timeout": "timeout_seconds",
            "max_instances": "max_instances",
            "coalesce": "coalesce",
            "misfire_grace_time": "misfire_grace_time",
            "tags": "tags",
            "metadata": "metadata",
        }

        for req_key, db_key in field_map.items():
            if req_key in body and body[req_key] is not None:
                update_data[db_key] = body[req_key]

        if "url" in body and body["url"] is not None:
            _check_ssrf(body["url"])

        if "auth" in body and body["auth"] is not None:
            update_data["auth_type"] = body["auth"].get("type", "none")
            update_data["auth_config"] = dict(body["auth"])

        if "retry" in body and body["retry"] is not None:
            r = body["retry"]
            update_data["max_retries"] = r.get("max_retries", 3)
            update_data["retry_delay_seconds"] = r.get("retry_delay", 1.0)
            update_data["retry_backoff"] = r.get("retry_backoff", 2.0)
            update_data["retry_max_delay_seconds"] = r.get("retry_max_delay", 60.0)
            update_data["retry_on_status"] = r.get("retry_on_status", [429, 500, 502, 503, 504])
            update_data["retry_on_timeout"] = r.get("retry_on_timeout", True)
            update_data["retry_on_failure"] = r.get("retry_on_failure", True)
            update_data["retry_jitter"] = r.get("retry_jitter", False)

        if "trigger" in body and body["trigger"] is not None:
            tc = dict(body["trigger"])
            update_data["trigger_type"] = tc.pop("type", job.get("trigger_type", "cron"))
            update_data["trigger_config"] = tc
            update_data["jitter_seconds"] = tc.get("jitter")
            update_data["start_date"] = tc.get("start_date")
            update_data["end_date"] = tc.get("end_date")

        await repo.update_job(job_id, update_data)
        updated = await repo.get_job_by_id(job_id)
        if updated is None:
            raise KeyError(f"Job '{job_id}' not found after update")

        self._reschedule_job(updated)

        logger.info("Job updated: %s", job_id)
        return updated

    async def delete_job(self, job_id: str) -> None:
        repo = self._get_repo()
        job = await repo.get_job_by_id(job_id)
        if not job:
            raise KeyError(f"Job '{job_id}' not found")

        self._remove_aps_job(job_id)
        await repo.update_job(job_id, {"status": JobStatus.DELETED.value})
        logger.info("Job deleted: %s", job_id)

    async def hard_delete_job(self, job_id: str) -> None:
        repo = self._get_repo()
        job = await repo.get_job_by_id(job_id)
        if not job:
            raise KeyError(f"Job '{job_id}' not found")

        self._remove_aps_job(job_id)
        await repo.delete_job(job_id)
        logger.info("Job hard deleted: %s", job_id)

    async def pause_job(self, job_id: str) -> dict[str, Any]:
        repo = self._get_repo()
        job = await repo.get_job_by_id(job_id)
        if not job:
            raise KeyError(f"Job '{job_id}' not found")
        if job.get("status") != JobStatus.ACTIVE.value:
            raise ValueError(f"Job is not active (current status: {job.get('status')})")

        self.pause_aps_job(job_id)
        await repo.update_job(job_id, {"status": JobStatus.PAUSED.value})
        updated = await repo.get_job_by_id(job_id)
        logger.info("Job paused: %s", job_id)
        return updated or job

    async def resume_job(self, job_id: str) -> dict[str, Any]:
        repo = self._get_repo()
        job = await repo.get_job_by_id(job_id)
        if not job:
            raise KeyError(f"Job '{job_id}' not found")
        if job.get("status") != JobStatus.PAUSED.value:
            raise ValueError(f"Job is not paused (current status: {job.get('status')})")

        self.resume_aps_job(job_id)
        await repo.update_job(job_id, {"status": JobStatus.ACTIVE.value})
        updated = await repo.get_job_by_id(job_id)
        logger.info("Job resumed: %s", job_id)
        return updated or job

    async def run_job_now(self, job_id: str) -> None:
        repo = self._get_repo()
        job = await repo.get_job_by_id(job_id)
        if not job:
            raise KeyError(f"Job '{job_id}' not found")
        if job.get("status") == JobStatus.DELETED.value:
            raise ValueError("Cannot run a deleted job")

        asyncio.create_task(self._execute_job(job_id, TriggerReason.MANUAL))
        logger.info("Job triggered manually: %s", job_id)

    async def list_jobs(
        self,
        status: str | None = None,
        tags: list[str] | None = None,
        page: int = 1,
        page_size: int = 20,
    ) -> tuple[list[dict[str, Any]], int]:
        repo = self._get_optional_repo()
        if repo is None:
            return [], 0
        return await repo.list_jobs(status=status, tags=tags, page=page, page_size=page_size)

    async def get_job(self, job_id: str) -> dict[str, Any] | None:
        repo = self._get_optional_repo()
        if repo is None:
            return None
        return await repo.get_job_by_id(job_id)

    async def get_job_history(
        self,
        job_id: str,
        page: int = 1,
        page_size: int = 20,
    ) -> tuple[list[dict[str, Any]], int]:
        repo = self._get_optional_repo()
        if repo is None:
            return [], 0
        return await repo.get_history(job_id=job_id, page=page, page_size=page_size)

    async def get_execution_history(
        self,
        page: int = 1,
        page_size: int = 50,
        status: str | None = None,
    ) -> tuple[list[dict[str, Any]], int]:
        repo = self._get_optional_repo()
        if repo is None:
            return [], 0
        return await repo.get_history_all(page=page, page_size=page_size, status_filter=status)

    async def get_metrics(self) -> dict[str, Any]:
        repo = self._get_optional_repo()
        if repo is None:
            return {"error": "Supabase not available"}
        total_jobs = await repo.count_jobs()
        active_jobs = await repo.count_jobs(status=JobStatus.ACTIVE.value)
        paused_jobs = await repo.count_jobs(status=JobStatus.PAUSED.value)
        total_executions = await repo.count_executions()
        successful_executions = await repo.count_executions(status=ExecutionStatus.SUCCESS.value)

        scheduler_status = self.get_scheduler_status()

        return {
            "jobs": {
                "total": total_jobs,
                "active": active_jobs,
                "paused": paused_jobs,
            },
            "executions": {
                "total": total_executions,
                "success": successful_executions,
                "failure": total_executions - successful_executions,
                "success_rate": round(successful_executions / total_executions * 100, 2) if total_executions else 0,
            },
            "scheduler": scheduler_status,
        }


# ---------------------------------------------------------------------------
# Validation helpers
# ---------------------------------------------------------------------------

def validate_cron_expression(expression: str) -> None:
    try:
        croniter(expression)
    except (ValueError, KeyError) as exc:
        raise ValueError(f"Invalid cron expression '{expression}': {exc}") from exc


def validate_timezone(tz: str) -> None:
    try:
        import pytz
        pytz.timezone(tz)
    except Exception as exc:
        raise ValueError(f"Unknown timezone: '{tz}'") from exc


def validate_url(url: str) -> str:
    if not url.startswith(("http://", "https://")):
        raise ValueError("Only http/https URLs are supported")
    return url


def job_to_response(job: dict[str, Any]) -> dict[str, Any]:
    return {
        "id": job.get("id"),
        "name": job.get("name"),
        "description": job.get("description"),
        "url": job.get("url"),
        "method": job.get("method", "GET"),
        "headers": job.get("headers", {}),
        "query_params": job.get("query_params", {}),
        "body": job.get("body"),
        "body_type": job.get("body_type", "json"),
        "auth_type": job.get("auth_type", "none"),
        "trigger_type": job.get("trigger_type"),
        "trigger_config": job.get("trigger_config", {}),
        "timezone": job.get("timezone", "UTC"),
        "timeout": job.get("timeout_seconds", 30.0),
        "max_retries": job.get("max_retries", 3),
        "retry_delay": job.get("retry_delay_seconds", 1.0),
        "retry_backoff": job.get("retry_backoff", 2.0),
        "retry_max_delay": job.get("retry_max_delay_seconds", 60.0),
        "retry_on_status": job.get("retry_on_status", [429, 500, 502, 503, 504]),
        "retry_on_timeout": job.get("retry_on_timeout", True),
        "retry_on_failure": job.get("retry_on_failure", True),
        "retry_jitter": job.get("retry_jitter", False),
        "max_instances": job.get("max_instances", 1),
        "coalesce": job.get("coalesce", True),
        "misfire_grace_time": job.get("misfire_grace_time"),
        "jitter": job.get("jitter_seconds"),
        "start_date": job.get("start_date"),
        "end_date": job.get("end_date"),
        "status": job.get("status", "active"),
        "tags": job.get("tags", []),
        "metadata": job.get("metadata", {}),
        "created_at": job.get("created_at"),
        "updated_at": job.get("updated_at"),
        "last_run_at": job.get("last_run_at"),
        "next_run_at": job.get("next_run_at"),
        "run_count": job.get("run_count", 0),
        "failure_count": job.get("failure_count", 0),
        "success_count": job.get("success_count", 0),
    }


def history_to_response(h: dict[str, Any]) -> dict[str, Any]:
    return {
        "id": h.get("id"),
        "job_id": h.get("job_id"),
        "job_name": h.get("job_name"),
        "started_at": h.get("started_at"),
        "ended_at": h.get("ended_at"),
        "duration_ms": h.get("duration_ms"),
        "status": h.get("status"),
        "http_status_code": h.get("http_status_code"),
        "response_size_bytes": h.get("response_size_bytes"),
        "retry_count": h.get("retry_count", 0),
        "trigger_reason": h.get("trigger_reason", "scheduled"),
        "error_message": h.get("error_message"),
        "exception_type": h.get("exception_type"),
        "request_url": h.get("request_url"),
        "request_method": h.get("request_method"),
        "response_preview": h.get("response_preview"),
        "created_at": h.get("created_at"),
    }