feat: cron job api added

.gitignore

@@ -134,4 +134,6 @@ tests
 deploy_hf.py
 .mimocode
 
-ddl
+ddl
+API_DESCRIPTION.md
+ENTERPRISE-API-ROADMAP*.md

app/api/deps.py

@@ -13,6 +13,7 @@ from app.services.extraction_service import ExtractionService
 from app.services.ocr_service import OCRService
 from app.services.sql_validator_service import SqlValidatorService
 from app.services.text_cleaner_service import TextCleanerService
+from app.services.scheduler_service import SchedulerService
 from app.services.vector_store_service import VectorStoreService
 
 
@@ -54,6 +55,11 @@ def get_vector_store_service() -> VectorStoreService:
     return _vector_store_service
 
 
+def get_scheduler_service() -> SchedulerService:
+    from app.api.server import _scheduler_service
+    return _scheduler_service
+
+
 def require_auth(token: str = Depends(require_api_key)) -> str:
     return token
 

app/api/server.py

@@ -16,6 +16,7 @@ from app.core.logger import get_logger
 from app.core.redis_client import close_redis, create_redis_client
 from app.core.scripts import load_scripts
 from app.services.embeddings_service import EmbeddingService
+from app.services.scheduler_service import SchedulerService
 from app.services.vector_store_service import VectorStoreService
 
 _logger = get_logger(__name__)
@@ -23,6 +24,7 @@ _settings = get_settings()
 
 _embedding_service: EmbeddingService = EmbeddingService()
 _vector_store_service: VectorStoreService = VectorStoreService(_embedding_service)
+_scheduler_service: SchedulerService = SchedulerService()
 
 
 async def _self_ping():
@@ -78,8 +80,13 @@ async def lifespan(app: FastAPI):
         _logger.warning("Redis not configured, running in degraded mode")
 
     asyncio.create_task(_self_ping())
+
+    await _scheduler_service.start()
+    _logger.info("Scheduler service started")
+
     yield
     _logger.info("Shutting down...")
+    await _scheduler_service.shutdown()
     await close_redis(redis)
     await _vector_store_service.close_all()
     await pool_manager.close_all()

app/api/v1/router.py

@@ -16,6 +16,7 @@ from app.api.v1 import (
     qr_decoder,
     qr_generator,
     reconcile,
+    scheduler,
     scraper,
     semantic_router,
     sql_validator,
@@ -39,6 +40,7 @@ api_v1_router.include_router(embeddings.router, tags=["Embeddings"])
 api_v1_router.include_router(code_executor.router, tags=["Code Executor"])
 api_v1_router.include_router(verify_router, prefix="/verify", tags=["Verify"])
 api_v1_router.include_router(reconcile.router, tags=["Reconcile"])
+api_v1_router.include_router(scheduler.router, tags=["Scheduler"])
 api_v1_router.include_router(scraper.router, tags=["Web Scraping"])
 api_v1_router.include_router(web_search.router, tags=["Web Search"])
 api_v1_router.include_router(sql_validator.router, tags=["SQL Validator"])

app/api/v1/scheduler.py

@@ -0,0 +1,326 @@
+from __future__ import annotations
+
+from typing import Any
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+
+from app.api.deps import get_scheduler_service, require_auth
+from app.core.logger import get_logger
+from app.services.scheduler_service import (
+    SchedulerService,
+    job_to_response,
+    history_to_response,
+    validate_cron_expression,
+    validate_timezone,
+    validate_url,
+)
+
+router = APIRouter()
+logger = get_logger(__name__)
+
+
+def _paginate(items: list[Any], total: int, page: int, page_size: int) -> dict[str, Any]:
+    return {
+        "items": items,
+        "total": total,
+        "page": page,
+        "page_size": page_size,
+        "has_next": (page * page_size) < total,
+        "has_prev": page > 1,
+    }
+
+
+# ---------------------------------------------------------------------------
+# Scheduler Monitoring
+# ---------------------------------------------------------------------------
+
+@router.get("/scheduler/status", summary="Get scheduler status")
+async def get_scheduler_status(
+    token: str = Depends(require_auth),
+    scheduler_service: SchedulerService = Depends(get_scheduler_service),
+):
+    status = scheduler_service.get_scheduler_status()
+    status["running_job_ids"] = scheduler_service.get_running_job_ids()
+    return {"success": True, "data": status}
+
+
+@router.get("/scheduler/metrics", summary="Get scheduler metrics")
+async def get_scheduler_metrics(
+    token: str = Depends(require_auth),
+    scheduler_service: SchedulerService = Depends(get_scheduler_service),
+):
+    metrics = await scheduler_service.get_metrics()
+    return {"success": True, "data": metrics}
+
+
+@router.get("/scheduler/health", summary="Get scheduler health (Redis + instance status)")
+async def get_scheduler_health(
+    scheduler_service: SchedulerService = Depends(get_scheduler_service),
+):
+    health = await scheduler_service.get_health()
+    return {"success": True, "data": health}
+
+
+# ---------------------------------------------------------------------------
+# Job CRUD
+# ---------------------------------------------------------------------------
+
+@router.post("/scheduler/jobs", summary="Create a new scheduled job", status_code=201)
+async def create_job(
+    body: dict[str, Any],
+    token: str = Depends(require_auth),
+    scheduler_service: SchedulerService = Depends(get_scheduler_service),
+):
+    errors = _validate_job_create(body)
+    if errors:
+        raise HTTPException(status_code=422, detail={"success": False, "errors": errors})
+
+    try:
+        job = await scheduler_service.create_job(body)
+        return {"success": True, "data": job_to_response(job)}
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail={"success": False, "error": str(exc)}) from exc
+
+
+@router.get("/scheduler/jobs", summary="List all scheduled jobs")
+async def list_jobs(
+    status: str | None = Query(None, description="Filter by status (active, paused, completed, failed, deleted)"),
+    tags: str | None = Query(None, description="Comma-separated tags to filter by"),
+    page: int = Query(1, ge=1, description="Page number"),
+    page_size: int = Query(20, ge=1, le=100, description="Items per page"),
+    token: str = Depends(require_auth),
+    scheduler_service: SchedulerService = Depends(get_scheduler_service),
+):
+    tag_list = [t.strip() for t in tags.split(",") if t.strip()] if tags else None
+    jobs, total = await scheduler_service.list_jobs(
+        status=status, tags=tag_list, page=page, page_size=page_size,
+    )
+    return {
+        "success": True,
+        "data": _paginate(
+            items=[job_to_response(j) for j in jobs],
+            total=total, page=page, page_size=page_size,
+        ),
+    }
+
+
+@router.get("/scheduler/jobs/{job_id}", summary="Get a scheduled job by ID")
+async def get_job(
+    job_id: str,
+    token: str = Depends(require_auth),
+    scheduler_service: SchedulerService = Depends(get_scheduler_service),
+):
+    job = await scheduler_service.get_job(job_id)
+    if not job or job.get("status") == "deleted":
+        raise HTTPException(status_code=404, detail={"success": False, "error": "Job not found"})
+    return {"success": True, "data": job_to_response(job)}
+
+
+@router.put("/scheduler/jobs/{job_id}", summary="Update a scheduled job")
+async def update_job(
+    job_id: str,
+    body: dict[str, Any],
+    token: str = Depends(require_auth),
+    scheduler_service: SchedulerService = Depends(get_scheduler_service),
+):
+    try:
+        job = await scheduler_service.update_job(job_id, body)
+        return {"success": True, "data": job_to_response(job)}
+    except KeyError as exc:
+        raise HTTPException(status_code=404, detail={"success": False, "error": str(exc)}) from exc
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail={"success": False, "error": str(exc)}) from exc
+
+
+@router.delete("/scheduler/jobs/{job_id}", summary="Delete a scheduled job (soft delete)")
+async def delete_job(
+    job_id: str,
+    token: str = Depends(require_auth),
+    scheduler_service: SchedulerService = Depends(get_scheduler_service),
+):
+    try:
+        await scheduler_service.delete_job(job_id)
+        return {"success": True, "message": "Job deleted successfully"}
+    except KeyError as exc:
+        raise HTTPException(status_code=404, detail={"success": False, "error": str(exc)}) from exc
+
+
+@router.delete("/scheduler/jobs/{job_id}/hard", summary="Permanently delete a scheduled job")
+async def hard_delete_job(
+    job_id: str,
+    token: str = Depends(require_auth),
+    scheduler_service: SchedulerService = Depends(get_scheduler_service),
+):
+    try:
+        await scheduler_service.hard_delete_job(job_id)
+        return {"success": True, "message": "Job permanently deleted"}
+    except KeyError as exc:
+        raise HTTPException(status_code=404, detail={"success": False, "error": str(exc)}) from exc
+
+
+# ---------------------------------------------------------------------------
+# Job Control
+# ---------------------------------------------------------------------------
+
+@router.post("/scheduler/jobs/{job_id}/pause", summary="Pause a scheduled job")
+async def pause_job(
+    job_id: str,
+    token: str = Depends(require_auth),
+    scheduler_service: SchedulerService = Depends(get_scheduler_service),
+):
+    try:
+        job = await scheduler_service.pause_job(job_id)
+        return {"success": True, "data": job_to_response(job)}
+    except KeyError as exc:
+        raise HTTPException(status_code=404, detail={"success": False, "error": str(exc)}) from exc
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail={"success": False, "error": str(exc)}) from exc
+
+
+@router.post("/scheduler/jobs/{job_id}/resume", summary="Resume a paused job")
+async def resume_job(
+    job_id: str,
+    token: str = Depends(require_auth),
+    scheduler_service: SchedulerService = Depends(get_scheduler_service),
+):
+    try:
+        job = await scheduler_service.resume_job(job_id)
+        return {"success": True, "data": job_to_response(job)}
+    except KeyError as exc:
+        raise HTTPException(status_code=404, detail={"success": False, "error": str(exc)}) from exc
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail={"success": False, "error": str(exc)}) from exc
+
+
+@router.post("/scheduler/jobs/{job_id}/run", summary="Trigger a job execution immediately")
+async def run_job_now(
+    job_id: str,
+    token: str = Depends(require_auth),
+    scheduler_service: SchedulerService = Depends(get_scheduler_service),
+):
+    try:
+        await scheduler_service.run_job_now(job_id)
+        return {"success": True, "message": "Job execution triggered"}
+    except KeyError as exc:
+        raise HTTPException(status_code=404, detail={"success": False, "error": str(exc)}) from exc
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail={"success": False, "error": str(exc)}) from exc
+
+
+# ---------------------------------------------------------------------------
+# Execution History
+# ---------------------------------------------------------------------------
+
+@router.get("/scheduler/jobs/{job_id}/history", summary="Get execution history for a job")
+async def get_job_history(
+    job_id: str,
+    page: int = Query(1, ge=1, description="Page number"),
+    page_size: int = Query(20, ge=1, le=100, description="Items per page"),
+    token: str = Depends(require_auth),
+    scheduler_service: SchedulerService = Depends(get_scheduler_service),
+):
+    job = await scheduler_service.get_job(job_id)
+    if not job:
+        raise HTTPException(status_code=404, detail={"success": False, "error": "Job not found"})
+
+    history, total = await scheduler_service.get_job_history(
+        job_id=job_id, page=page, page_size=page_size,
+    )
+    return {
+        "success": True,
+        "data": _paginate(
+            items=[history_to_response(h) for h in history],
+            total=total, page=page, page_size=page_size,
+        ),
+    }
+
+
+@router.get("/scheduler/history", summary="Get global execution history")
+async def get_execution_history(
+    page: int = Query(1, ge=1, description="Page number"),
+    page_size: int = Query(50, ge=1, le=200, description="Items per page"),
+    status: str | None = Query(None, description="Filter by execution status"),
+    token: str = Depends(require_auth),
+    scheduler_service: SchedulerService = Depends(get_scheduler_service),
+):
+    history, total = await scheduler_service.get_execution_history(
+        page=page, page_size=page_size, status=status,
+    )
+    return {
+        "success": True,
+        "data": _paginate(
+            items=[history_to_response(h) for h in history],
+            total=total, page=page, page_size=page_size,
+        ),
+    }
+
+
+# ---------------------------------------------------------------------------
+# Validation
+# ---------------------------------------------------------------------------
+
+def _validate_job_create(body: dict[str, Any]) -> list[str]:
+    errors: list[str] = []
+
+    if not body.get("name"):
+        errors.append("name is required")
+    elif not isinstance(body["name"], str) or len(body["name"]) > 255:
+        errors.append("name must be a string between 1 and 255 characters")
+
+    if not body.get("url"):
+        errors.append("url is required")
+    elif not isinstance(body["url"], str):
+        errors.append("url must be a string")
+    else:
+        try:
+            validate_url(body["url"])
+        except ValueError as exc:
+            errors.append(str(exc))
+
+    trigger = body.get("trigger")
+    if not trigger:
+        errors.append("trigger is required")
+    elif not isinstance(trigger, dict):
+        errors.append("trigger must be an object")
+    else:
+        trigger_type = trigger.get("type")
+        if not trigger_type:
+            errors.append("trigger.type is required")
+        elif trigger_type not in ("cron", "interval", "date"):
+            errors.append("trigger.type must be one of: cron, interval, date")
+        elif trigger_type == "cron":
+            if not trigger.get("cron_expression"):
+                errors.append("trigger.cron_expression is required for cron trigger")
+            else:
+                try:
+                    validate_cron_expression(trigger["cron_expression"])
+                except ValueError as exc:
+                    errors.append(str(exc))
+
+    if body.get("timezone"):
+        try:
+            validate_timezone(body["timezone"])
+        except ValueError as exc:
+            errors.append(str(exc))
+
+    timeout = body.get("timeout")
+    if timeout is not None:
+        if not isinstance(timeout, (int, float)):
+            errors.append("timeout must be a number")
+        elif timeout < 1 or timeout > 300:
+            errors.append("timeout must be between 1 and 300")
+
+    method = body.get("method")
+    if method and method not in ("GET", "POST", "PUT", "PATCH", "DELETE"):
+        errors.append("method must be one of: GET, POST, PUT, PATCH, DELETE")
+
+    retry = body.get("retry")
+    if retry is not None:
+        if not isinstance(retry, dict):
+            errors.append("retry must be an object")
+        else:
+            max_retries = retry.get("max_retries")
+            if max_retries is not None and (not isinstance(max_retries, int) or max_retries < 0 or max_retries > 10):
+                errors.append("retry.max_retries must be between 0 and 10")
+
+    return errors

app/config.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 from functools import lru_cache
 from typing import Optional
 
+from pydantic import Field
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 
@@ -83,6 +84,27 @@ class Settings(BaseSettings):
     lockout_minutes: int = 15
     admin_password: str = ""
 
+    # Redis connection
+    redis_host: str = Field(default="localhost", alias="REDIS_HOST")
+    redis_port: int = Field(default=6379, alias="REDIS_PORT")
+    redis_db: int = Field(default=0, alias="REDIS_DB")
+    redis_password: str | None = Field(default=None, alias="REDIS_PASSWORD")
+    redis_ssl: bool = Field(default=False, alias="REDIS_SSL")
+    redis_socket_timeout: int = Field(default=30, alias="REDIS_SOCKET_TIMEOUT")
+    redis_socket_connect_timeout: int = Field(default=30, alias="REDIS_SOCKET_CONNECT_TIMEOUT")
+    redis_retry_on_timeout: bool = Field(default=True, alias="REDIS_RETRY_ON_TIMEOUT")
+    redis_health_check_interval: int = Field(default=30, alias="REDIS_HEALTH_CHECK_INTERVAL")
+
+    # Scheduler settings
+    max_http_timeout: float = 300.0
+    default_scheduler_timezone: str = "UTC"
+    history_retention_days: int = 30
+    ssrf_protection: bool = True
+    scheduler_instance_id: str | None = Field(default=None, alias="SCHEDULER_INSTANCE_ID")
+    scheduler_lock_timeout: int = Field(default=300, alias="SCHEDULER_LOCK_TIMEOUT")
+    scheduler_misfire_grace_time: int = Field(default=300, alias="SCHEDULER_MISFIRE_GRACE_TIME")
+    scheduler_coordinator_prefix: str = Field(default="scheduler:", alias="SCHEDULER_COORDINATOR_PREFIX")
+
     @property
     def max_upload_mb(self) -> int:
         return self.max_upload_bytes // (1024 * 1024)

app/services/scheduler_service.py

@@ -0,0 +1,1358 @@
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import json
+import re
+import secrets
+import uuid
+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
+
+logger = get_logger(__name__)
+settings = get_settings()
+
+# ---------------------------------------------------------------------------
+# 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._client: httpx.AsyncClient | None = None
+        self._lock = asyncio.Lock()
+
+    async def _get_client(self) -> httpx.AsyncClient:
+        if self._client is None or self._client.is_closed:
+            async with self._lock:
+                if self._client is None or self._client.is_closed:
+                    self._client = httpx.AsyncClient(
+                        follow_redirects=True,
+                        limits=httpx.Limits(
+                            max_connections=200,
+                            max_keepalive_connections=50,
+                            keepalive_expiry=30,
+                        ),
+                        timeout=httpx.Timeout(settings.max_http_timeout),
+                    )
+        return self._client
+
+    async def close(self) -> None:
+        if self._client and not self._client.is_closed:
+            await self._client.aclose()
+            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 as exc:
+            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
+
+    # -----------------------------------------------------------------------
+    # Redis connection management
+    # -----------------------------------------------------------------------
+
+    async def _connect_async_redis(self) -> Any:
+        ra = _get_redis_asyncio()
+        if ra is None:
+            return None
+        try:
+            client = ra.Redis(
+                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,
+                decode_responses=True,
+            )
+            await client.ping()
+            logger.info(
+                "Connected to Redis at %s:%d (db=%d)",
+                settings.redis_host, settings.redis_port, settings.redis_db,
+            )
+            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(
+            db=settings.redis_db,
+            host=settings.redis_host,
+            port=settings.redis_port,
+            password=settings.redis_password or None,
+            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,
+        )
+
+    # -----------------------------------------------------------------------
+    # 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:
+            client = get_supabase_client()
+            if client:
+                repo = SchedulerRepository(client)
+                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")
+                client = get_supabase_client()
+                if client:
+                    repo = SchedulerRepository(client)
+                    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,
+        })
+
+        client = get_supabase_client()
+        if client is None:
+            logger.error("Supabase not available, cannot execute job %s", job_id)
+            return
+
+        repo = SchedulerRepository(client)
+        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
+
+        headers_hash = hashlib.sha256(
+            json.dumps(job.get("headers") or {}, sort_keys=True).encode()
+        ).hexdigest() if job.get("headers") else ""
+
+        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]:
+        client = get_supabase_client()
+        if client is None:
+            raise RuntimeError("Supabase not available")
+
+        repo = SchedulerRepository(client)
+
+        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]:
+        client = get_supabase_client()
+        if client is None:
+            raise RuntimeError("Supabase not available")
+
+        repo = SchedulerRepository(client)
+        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:
+        client = get_supabase_client()
+        if client is None:
+            raise RuntimeError("Supabase not available")
+
+        repo = SchedulerRepository(client)
+        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:
+        client = get_supabase_client()
+        if client is None:
+            raise RuntimeError("Supabase not available")
+
+        repo = SchedulerRepository(client)
+        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]:
+        client = get_supabase_client()
+        if client is None:
+            raise RuntimeError("Supabase not available")
+
+        repo = SchedulerRepository(client)
+        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]:
+        client = get_supabase_client()
+        if client is None:
+            raise RuntimeError("Supabase not available")
+
+        repo = SchedulerRepository(client)
+        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:
+        client = get_supabase_client()
+        if client is None:
+            raise RuntimeError("Supabase not available")
+
+        repo = SchedulerRepository(client)
+        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]:
+        client = get_supabase_client()
+        if client is None:
+            return [], 0
+        repo = SchedulerRepository(client)
+        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:
+        client = get_supabase_client()
+        if client is None:
+            return None
+        repo = SchedulerRepository(client)
+        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]:
+        client = get_supabase_client()
+        if client is None:
+            return [], 0
+        repo = SchedulerRepository(client)
+        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]:
+        client = get_supabase_client()
+        if client is None:
+            return [], 0
+        repo = SchedulerRepository(client)
+        return await repo.get_history_all(page=page, page_size=page_size, status_filter=status)
+
+    async def get_metrics(self) -> dict[str, Any]:
+        client = get_supabase_client()
+        if client is None:
+            return {"error": "Supabase not available"}
+
+        repo = SchedulerRepository(client)
+        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"),
+    }

pyproject.toml

@@ -33,6 +33,10 @@ dependencies = [
     "phonenumbers>=8.13.0",
     "jsonschema>=4.21.0",
     "supabase>=2.0.0",
+    "apscheduler>=3.10.4",
+    "croniter>=2.0.0",
+    "pytz>=2024.1",
+    "redis>=5.0.0",
 ]
 
 [project.optional-dependencies]

requirements.txt

@@ -67,3 +67,8 @@ jsonpath-ng>=1.7.0
 
 # QR code generation (testing)
 qrcode[pil]>=8.0
+
+# Scheduler
+apscheduler>=3.10.4
+croniter>=2.0.0
+pytz>=2024.1