appfactory-portal/app/db/jobs.py

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 retry_failed_job(job_id: int, user: dict[str, Any]):
    run_migrations()
    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()
    source_job = con.execute(
        """
        SELECT id
        FROM jobs
        WHERE id = ?
          AND status = 'failed'
        """,
        (job_id,),
    ).fetchone()
    if not source_job:
        con.close()
        return None

    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
        )
        SELECT
            type,
            target_type,
            target_id,
            payload_json,
            'queued',
            ?,
            ?,
            ?,
            'portal_retry'
        FROM jobs
        WHERE id = ?
          AND status = 'failed'
        """,
        (created_by_user_id, created_by_username, created_by_display_name, job_id),
    )
    if cur.rowcount != 1:
        con.close()
        return None

    new_job_id = cur.lastrowid
    con.commit()
    con.close()

    append_job_log(new_job_id, "system", f"Job queued as retry of #{job_id}.")
    return new_job_id


def cancel_job(job_id: int):
    run_migrations()
    con = get_connection()

    row = con.execute(
        """
        SELECT status
        FROM jobs
        WHERE id = ?
        """,
        (job_id,),
    ).fetchone()
    if not row:
        con.close()
        return None

    status = (row["status"] or "").lower()
    if status == "queued":
        new_status = "cancelled"
        cur = con.execute(
            """
            UPDATE jobs
            SET status = 'cancelled',
                finished_at = CURRENT_TIMESTAMP
            WHERE id = ?
              AND status = 'queued'
            """,
            (job_id,),
        )
    elif status == "running":
        new_status = "cancelled_requested"
        cur = con.execute(
            """
            UPDATE jobs
            SET status = 'cancelled_requested'
            WHERE id = ?
              AND status = 'running'
            """,
            (job_id,),
        )
    else:
        con.close()
        return None

    if cur.rowcount != 1:
        con.close()
        return None

    con.commit()
    con.close()
    append_job_log(job_id, "system", f"Job status changed from {status} to {new_status}.")
    return {"previous_status": status, "status": new_status}


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,
    offset: int = 0,
    status: str | None = None,
    job_type: str | None = None,
    target: str | None = None,
):
    run_migrations()
    con = get_connection()

    filters = []
    params: list[Any] = []
    if status:
        filters.append("status = ?")
        params.append(status)
    if job_type:
        filters.append("type = ?")
        params.append(job_type)
    if target:
        filters.append("(target_type LIKE ? OR target_id LIKE ?)")
        target_pattern = f"%{target}%"
        params.extend([target_pattern, target_pattern])

    where_sql = f"WHERE {' AND '.join(filters)}" if filters else ""
    rows = con.execute(
        f"""
        SELECT *
        FROM jobs
        {where_sql}
        ORDER BY
            CASE
                WHEN status = 'running' THEN 0
                WHEN status = 'queued' THEN 1
                ELSE 2
            END,
            id DESC
        LIMIT ?
        OFFSET ?
        """,
        (*params, limit, offset),
    ).fetchall()

    con.close()
    return [dict(row) for row in rows]


def count_jobs(
    status: str | None = None,
    job_type: str | None = None,
    target: str | None = None,
):
    run_migrations()
    con = get_connection()

    filters = []
    params: list[Any] = []
    if status:
        filters.append("status = ?")
        params.append(status)
    if job_type:
        filters.append("type = ?")
        params.append(job_type)
    if target:
        filters.append("(target_type LIKE ? OR target_id LIKE ?)")
        target_pattern = f"%{target}%"
        params.extend([target_pattern, target_pattern])

    where_sql = f"WHERE {' AND '.join(filters)}" if filters else ""
    row = con.execute(
        f"""
        SELECT COUNT(*) AS count
        FROM jobs
        {where_sql}
        """,
        params,
    ).fetchone()

    con.close()
    return row["count"] if row else 0


def get_job_stats():
    run_migrations()
    con = get_connection()

    rows = con.execute(
        """
        SELECT
            SUM(CASE WHEN status = 'queued' THEN 1 ELSE 0 END) AS queued,
            SUM(CASE WHEN status = 'running' THEN 1 ELSE 0 END) AS running,
            SUM(CASE WHEN status = 'failed' THEN 1 ELSE 0 END) AS failed,
            SUM(CASE WHEN status = 'success' THEN 1 ELSE 0 END) AS success,
            SUM(CASE WHEN status = 'failed' AND created_at >= datetime('now', '-24 hours') THEN 1 ELSE 0 END) AS failed_24h,
            SUM(CASE WHEN status = 'success' AND created_at >= datetime('now', '-24 hours') THEN 1 ELSE 0 END) AS success_24h
        FROM jobs
        """
    ).fetchone()

    con.close()
    return dict(rows) if rows else {
        "queued": 0,
        "running": 0,
        "failed": 0,
        "success": 0,
        "failed_24h": 0,
        "success_24h": 0,
    }


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 get_job_logs_after(job_id: int, last_log_id: int = 0, limit: int = 100):
    run_migrations()
    con = get_connection()

    rows = con.execute(
        """
        SELECT *
        FROM job_logs
        WHERE job_id = ?
          AND id > ?
        ORDER BY id
        LIMIT ?
        """,
        (job_id, last_log_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