Add scheduled scripts scheduler

app/main.py

@@ -1,7 +1,9 @@
+import json
 import os
 import sqlite3
 import threading
 import time
+from datetime import datetime, timedelta
 from pathlib import Path
 
 import requests
@@ -12,6 +14,7 @@ DB_FILE = Path("/opt/appfactory/data/appfactory/appfactory.db")
 CHECK_INTERVAL_SECONDS = int(os.getenv("APPFACTORY_MONITOR_INTERVAL_SECONDS", "30"))
 REQUEST_TIMEOUT_SECONDS = int(os.getenv("APPFACTORY_MONITOR_TIMEOUT_SECONDS", "5"))
 HEALTH_RETENTION_DAYS = int(os.getenv("APPFACTORY_HEALTH_RETENTION_DAYS", "7"))
+SCHEDULER_INTERVAL_SECONDS = int(os.getenv("APPFACTORY_SCHEDULER_INTERVAL_SECONDS", "60"))
 
 CORE_SERVICES = {
     "appfactory-portal": "http://appfactory-portal:9100/health",
@@ -20,9 +23,12 @@ CORE_SERVICES = {
     "appfactory-monitor": "http://appfactory-monitor:9300/health",
 }
 
+TERMINAL_JOB_STATUSES = {"success", "failed", "cancelled"}
+
 app = FastAPI(title="AppFactory Monitor")
 
 _monitor_thread = None
+_scheduler_thread = None
 _monitor_started = False
 
 
@@ -185,7 +191,13 @@ def open_incident(service_id: str, health_id: int, status: str, error_text: str
         """,
         (
             service_id,
-            f'{{"health_id": {health_id}, "status": "{status}"}}',
+            json.dumps(
+                {
+                    "health_id": health_id,
+                    "status": status,
+                },
+                ensure_ascii=False,
+            ),
         ),
     )
 
@@ -242,7 +254,13 @@ def resolve_incident(service_id: str, health_id: int):
         """,
         (
             service_id,
-            f'{{"incident_id": {incident["id"]}, "health_id": {health_id}}}',
+            json.dumps(
+                {
+                    "incident_id": incident["id"],
+                    "health_id": health_id,
+                },
+                ensure_ascii=False,
+            ),
         ),
     )
 
@@ -332,6 +350,228 @@ def run_check_cycle():
         check_service(service_id, url)
 
 
+def parse_time_hhmm(value: str | None) -> tuple[int, int]:
+    if not value:
+        return 3, 0
+
+    parts = value.strip().split(":", 1)
+
+    if len(parts) != 2:
+        return 3, 0
+
+    try:
+        hour = int(parts[0])
+        minute = int(parts[1])
+    except ValueError:
+        return 3, 0
+
+    if hour < 0 or hour > 23:
+        hour = 3
+
+    if minute < 0 or minute > 59:
+        minute = 0
+
+    return hour, minute
+
+
+def compute_next_run_at(schedule_type: str, schedule_time: str | None) -> str:
+    now = datetime.utcnow().replace(microsecond=0)
+    schedule_type = (schedule_type or "daily").strip().lower()
+    hour, minute = parse_time_hhmm(schedule_time)
+
+    if schedule_type == "hourly":
+        candidate = now.replace(minute=minute, second=0)
+        if candidate <= now:
+            candidate += timedelta(hours=1)
+        return candidate.strftime("%Y-%m-%d %H:%M:%S")
+
+    if schedule_type == "weekly":
+        candidate = now.replace(hour=hour, minute=minute, second=0)
+        while candidate <= now:
+            candidate += timedelta(days=7)
+        return candidate.strftime("%Y-%m-%d %H:%M:%S")
+
+    if schedule_type == "monthly":
+        candidate = now.replace(day=1, hour=hour, minute=minute, second=0)
+        if candidate <= now:
+            if candidate.month == 12:
+                candidate = candidate.replace(year=candidate.year + 1, month=1)
+            else:
+                candidate = candidate.replace(month=candidate.month + 1)
+        return candidate.strftime("%Y-%m-%d %H:%M:%S")
+
+    candidate = now.replace(hour=hour, minute=minute, second=0)
+    if candidate <= now:
+        candidate += timedelta(days=1)
+
+    return candidate.strftime("%Y-%m-%d %H:%M:%S")
+
+
+def sync_scheduled_script_results():
+    con = get_connection()
+
+    rows = con.execute(
+        """
+        SELECT
+            scheduled_scripts.id AS scheduled_script_id,
+            scheduled_scripts.last_job_id AS last_job_id,
+            jobs.status AS job_status,
+            jobs.error_text AS job_error_text,
+            jobs.started_at AS job_started_at,
+            jobs.finished_at AS job_finished_at
+        FROM scheduled_scripts
+        JOIN jobs ON jobs.id = scheduled_scripts.last_job_id
+        WHERE scheduled_scripts.is_running = 1
+          AND jobs.status IN ('success', 'failed', 'cancelled')
+        """
+    ).fetchall()
+
+    for row in rows:
+        last_run_at = row["job_finished_at"] or row["job_started_at"]
+
+        con.execute(
+            """
+            UPDATE scheduled_scripts
+            SET is_running = 0,
+                last_status = ?,
+                last_error_text = ?,
+                last_run_at = ?,
+                updated_at = CURRENT_TIMESTAMP
+            WHERE id = ?
+            """,
+            (
+                row["job_status"],
+                row["job_error_text"],
+                last_run_at,
+                row["scheduled_script_id"],
+            ),
+        )
+
+    con.commit()
+    con.close()
+
+
+def enqueue_due_scheduled_scripts():
+    con = get_connection()
+
+    con.execute("BEGIN IMMEDIATE")
+
+    rows = con.execute(
+        """
+        SELECT *
+        FROM scheduled_scripts
+        WHERE is_enabled = 1
+          AND is_running = 0
+          AND next_run_at IS NOT NULL
+          AND next_run_at <= CURRENT_TIMESTAMP
+        ORDER BY next_run_at ASC, id ASC
+        """
+    ).fetchall()
+
+    for row in rows:
+        script_id = int(row["id"])
+        script_name = row["script_name"]
+        timeout_seconds = int(row["timeout_seconds"] or 300)
+        next_run_at = compute_next_run_at(row["schedule_type"], row["schedule_time"])
+
+        payload = {
+            "scheduled_script_id": script_id,
+            "script_name": script_name,
+            "timeout_seconds": timeout_seconds,
+        }
+
+        cur = con.execute(
+            """
+            INSERT INTO jobs (
+                type,
+                target_type,
+                target_id,
+                status,
+                source,
+                created_by_username,
+                payload_json,
+                created_at
+            )
+            VALUES (
+                'run_script',
+                'scheduled_script',
+                ?,
+                'queued',
+                'scheduler',
+                'system',
+                ?,
+                CURRENT_TIMESTAMP
+            )
+            """,
+            (
+                script_name,
+                json.dumps(payload, ensure_ascii=False),
+            ),
+        )
+
+        job_id = int(cur.lastrowid)
+
+        con.execute(
+            """
+            UPDATE scheduled_scripts
+            SET is_running = 1,
+                last_job_id = ?,
+                last_status = 'queued',
+                last_error_text = NULL,
+                next_run_at = ?,
+                updated_at = CURRENT_TIMESTAMP
+            WHERE id = ?
+            """,
+            (
+                job_id,
+                next_run_at,
+                script_id,
+            ),
+        )
+
+        con.execute(
+            """
+            INSERT INTO audit_events (
+                username,
+                action,
+                target_type,
+                target_id,
+                source,
+                metadata,
+                created_at
+            )
+            VALUES (
+                'system',
+                'scheduled_script.enqueued',
+                'scheduled_script',
+                ?,
+                'scheduler',
+                ?,
+                CURRENT_TIMESTAMP
+            )
+            """,
+            (
+                str(script_id),
+                json.dumps(
+                    {
+                        "job_id": job_id,
+                        "script_name": script_name,
+                        "next_run_at": next_run_at,
+                    },
+                    ensure_ascii=False,
+                ),
+            ),
+        )
+
+    con.commit()
+    con.close()
+
+
+def run_scheduler_cycle():
+    sync_scheduled_script_results()
+    enqueue_due_scheduled_scripts()
+
+
 def monitor_loop():
     while True:
         try:
@@ -342,9 +582,20 @@ def monitor_loop():
         time.sleep(CHECK_INTERVAL_SECONDS)
 
 
+def scheduler_loop():
+    while True:
+        try:
+            run_scheduler_cycle()
+        except Exception as exc:
+            print(f"Scheduler loop error: {exc}", flush=True)
+
+        time.sleep(SCHEDULER_INTERVAL_SECONDS)
+
+
 @app.on_event("startup")
 def start_monitor():
     global _monitor_thread
+    global _scheduler_thread
     global _monitor_started
 
     if _monitor_started:
@@ -358,6 +609,12 @@ def start_monitor():
     )
     _monitor_thread.start()
 
+    _scheduler_thread = threading.Thread(
+        target=scheduler_loop,
+        daemon=True,
+    )
+    _scheduler_thread.start()
+
 
 @app.get("/health")
 def health():
@@ -366,6 +623,7 @@ def health():
         "interval_seconds": CHECK_INTERVAL_SECONDS,
         "timeout_seconds": REQUEST_TIMEOUT_SECONDS,
         "retention_days": HEALTH_RETENTION_DAYS,
+        "scheduler_interval_seconds": SCHEDULER_INTERVAL_SECONDS,
     }