import json from typing import Any from app.db.database import get_connection from app.db.migrations import run_migrations ACTIVE_STATUSES = ("queued", "running") def create_job( job_type: str, target_type: str, target_id: str, payload: dict[str, Any] | None = None, user: dict[str, Any] | None = None, source: str = "portal", ): run_migrations() payload_json = json.dumps(payload or {}, ensure_ascii=False, sort_keys=True) created_by_user_id = user.get("id") if user else None created_by_username = user.get("username") if user else None created_by_display_name = user.get("display_name") if user else None con = get_connection() cur = con.execute( """ INSERT INTO jobs ( type, target_type, target_id, payload_json, status, created_by_user_id, created_by_username, created_by_display_name, source ) VALUES (?, ?, ?, ?, 'queued', ?, ?, ?, ?) """, ( job_type, target_type, target_id, payload_json, created_by_user_id, created_by_username, created_by_display_name, source, ), ) job_id = cur.lastrowid con.commit() con.close() append_job_log(job_id, "system", "Job queued.") return job_id def update_job_status( job_id: int, status: str, worker_id: str | None = None, result: dict[str, Any] | None = None, error_text: str | None = None, ): run_migrations() result_json = json.dumps(result, ensure_ascii=False, sort_keys=True) if result is not None else None started_sql = ", started_at = COALESCE(started_at, CURRENT_TIMESTAMP)" if status == "running" else "" finished_sql = ", finished_at = CURRENT_TIMESTAMP" if status in {"success", "failed", "cancelled"} else "" con = get_connection() con.execute( f""" UPDATE jobs SET status = ?, worker_id = COALESCE(?, worker_id), result_json = COALESCE(?, result_json), error_text = COALESCE(?, error_text) {started_sql} {finished_sql} WHERE id = ? """, (status, worker_id, result_json, error_text, job_id), ) con.commit() con.close() def append_job_log(job_id: int, stream: str, message: str): run_migrations() con = get_connection() con.execute( """ INSERT INTO job_logs (job_id, stream, message) VALUES (?, ?, ?) """, (job_id, stream, message), ) con.commit() con.close() def get_next_queued_job(): run_migrations() con = get_connection() row = con.execute( """ SELECT * FROM jobs WHERE status = 'queued' AND NOT ( type IN ('deploy_app', 'deploy_core_service') AND EXISTS ( SELECT 1 FROM jobs active WHERE active.id != jobs.id AND active.type = jobs.type AND active.target_type = jobs.target_type AND active.target_id = jobs.target_id AND active.status = 'running' ) ) ORDER BY id LIMIT 1 """ ).fetchone() con.close() return dict(row) if row else None def get_jobs(limit: int = 100): run_migrations() con = get_connection() rows = con.execute( """ SELECT * FROM jobs ORDER BY CASE WHEN status = 'running' THEN 0 WHEN status = 'queued' THEN 1 ELSE 2 END, id DESC LIMIT ? """, (limit,), ).fetchall() con.close() return [dict(row) for row in rows] def get_job(job_id: int): run_migrations() con = get_connection() row = con.execute( """ SELECT * FROM jobs WHERE id = ? """, (job_id,), ).fetchone() con.close() return dict(row) if row else None def get_job_logs(job_id: int, limit: int = 1000): run_migrations() con = get_connection() rows = con.execute( """ SELECT * FROM job_logs WHERE job_id = ? ORDER BY id LIMIT ? """, (job_id, limit), ).fetchall() con.close() return [dict(row) for row in rows] def has_active_deploy_job(target_type: str, target_id: str): run_migrations() con = get_connection() row = con.execute( f""" SELECT id FROM jobs WHERE type IN ('deploy_app', 'deploy_core_service') AND target_type = ? AND target_id = ? AND status IN ({','.join('?' for _ in ACTIVE_STATUSES)}) ORDER BY id DESC LIMIT 1 """, (target_type, target_id, *ACTIVE_STATUSES), ).fetchone() con.close() return dict(row) if row else None