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"), }