appfactory-portal/app/db/apps.py

from app.db.database import get_connection


SUCCESS_STATUSES = ("ok", "success", "succeeded", "done", "deployed", "completed")
FAILED_STATUSES = ("failed", "failure", "error", "cancelled", "canceled")
RUNNING_STATUSES = ("running", "pending", "queued", "in_progress", "starting")


def get_apps():
    con = get_connection()

    rows = con.execute(
        """
        SELECT
            a.id,
            a.name,
            a.language,
            a.version,
            a.status,
            COALESCE(a.memory, '') AS memory,
            COALESCE(a.cpus, '') AS cpus,
            a.updated_at,
            (
                SELECT d.status
                FROM deployments d
                WHERE d.app_id = a.id
                ORDER BY d.id DESC
                LIMIT 1
            ) AS last_deploy_status,
            (
                SELECT d.started_at
                FROM deployments d
                WHERE d.app_id = a.id
                ORDER BY d.id DESC
                LIMIT 1
            ) AS last_deploy_started_at
        FROM apps a
        ORDER BY a.id
        """
    ).fetchall()

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


def get_app(app_id: str):
    con = get_connection()

    row = con.execute(
        """
        SELECT id, name, language, version, status, memory, cpus, updated_at
        FROM apps
        WHERE id = ?
        """,
        (app_id,),
    ).fetchone()

    con.close()
    return dict(row) if row else None


def update_app_resources(app_id: str, memory: str, cpus: str):
    con = get_connection()

    con.execute(
        """
        UPDATE apps
        SET memory = ?,
            cpus = ?,
            updated_at = CURRENT_TIMESTAMP
        WHERE id = ?
        """,
        (
            memory or None,
            cpus or None,
            app_id,
        ),
    )

    con.commit()
    con.close()


def build_deployment_filters(
    status: str | None = None,
    app_id: str | None = None,
    source: str | None = None,
):
    filters = []
    params = []

    if status:
        normalized_status = status.strip().lower()
        if normalized_status == "success":
            filters.append(f"LOWER(COALESCE(status, '')) IN ({','.join('?' for _ in SUCCESS_STATUSES)})")
            params.extend(SUCCESS_STATUSES)
        elif normalized_status == "failed":
            filters.append(f"LOWER(COALESCE(status, '')) IN ({','.join('?' for _ in FAILED_STATUSES)})")
            params.extend(FAILED_STATUSES)
        elif normalized_status == "running":
            filters.append(f"LOWER(COALESCE(status, '')) IN ({','.join('?' for _ in RUNNING_STATUSES)})")
            params.extend(RUNNING_STATUSES)
        else:
            filters.append("LOWER(status) = LOWER(?)")
            params.append(status)

    if app_id:
        filters.append("app_id = ?")
        params.append(app_id)

    if source:
        filters.append("trigger_source = ?")
        params.append(source)

    where = f"WHERE {' AND '.join(filters)}" if filters else ""
    return where, params


def get_deployments(
    limit: int = 20,
    offset: int = 0,
    status: str | None = None,
    app_id: str | None = None,
    source: str | None = None,
):
    con = get_connection()
    where, params = build_deployment_filters(status=status, app_id=app_id, source=source)

    rows = con.execute(
        f"""
        SELECT
            id,
            app_id,
            kind,
            status,
            started_at,
            finished_at,
            returncode,
            trigger_source,
            triggered_by_username,
            triggered_by_display_name
        FROM deployments
        {where}
        ORDER BY
            CASE
                WHEN LOWER(COALESCE(status, '')) IN ({','.join('?' for _ in RUNNING_STATUSES)}) THEN 0
                ELSE 1
            END,
            id DESC
        LIMIT ?
        OFFSET ?
        """,
        (*params, *RUNNING_STATUSES, limit, offset),
    ).fetchall()

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


def count_deployments(
    status: str | None = None,
    app_id: str | None = None,
    source: str | None = None,
):
    con = get_connection()
    where, params = build_deployment_filters(status=status, app_id=app_id, source=source)

    row = con.execute(
        f"""
        SELECT COUNT(*) AS count
        FROM deployments
        {where}
        """,
        params,
    ).fetchone()

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


def get_deployment_filter_options():
    con = get_connection()

    apps = con.execute(
        """
        SELECT DISTINCT app_id
        FROM deployments
        WHERE app_id IS NOT NULL AND app_id != ''
        ORDER BY app_id
        """
    ).fetchall()
    sources = con.execute(
        """
        SELECT DISTINCT trigger_source
        FROM deployments
        WHERE trigger_source IS NOT NULL AND trigger_source != ''
        ORDER BY trigger_source
        """
    ).fetchall()

    con.close()
    return {
        "apps": [row["app_id"] for row in apps],
        "sources": [row["trigger_source"] for row in sources],
    }


def get_deployment_stats():
    con = get_connection()

    row = con.execute(
        f"""
        SELECT
            COUNT(*) AS total,
            SUM(CASE WHEN LOWER(COALESCE(status, '')) IN ({','.join('?' for _ in FAILED_STATUSES)}) THEN 1 ELSE 0 END) AS failed,
            SUM(CASE WHEN LOWER(COALESCE(status, '')) IN ({','.join('?' for _ in RUNNING_STATUSES)}) THEN 1 ELSE 0 END) AS running,
            SUM(CASE WHEN LOWER(COALESCE(status, '')) IN ({','.join('?' for _ in SUCCESS_STATUSES)}) THEN 1 ELSE 0 END) AS success
        FROM deployments
        """,
        (*FAILED_STATUSES, *RUNNING_STATUSES, *SUCCESS_STATUSES),
    ).fetchone()

    con.close()
    stats = dict(row) if row else {"total": 0, "failed": 0, "running": 0, "success": 0}
    total = stats.get("total") or 0
    success = stats.get("success") or 0
    stats["success_rate"] = round((success / total) * 100, 1) if total else 0
    return stats


def get_app_deployments(app_id: str, limit: int = 10):
    con = get_connection()

    rows = con.execute(
        """
        SELECT
            id,
            app_id,
            kind,
            status,
            started_at,
            finished_at,
            returncode,
            trigger_source,
            triggered_by_username,
            triggered_by_display_name
        FROM deployments
        WHERE app_id = ?
        ORDER BY id DESC
        LIMIT ?
        """,
        (app_id, limit),
    ).fetchall()

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


def get_deployment(deployment_id: int):
    con = get_connection()

    row = con.execute(
        """
        SELECT
            id,
            app_id,
            kind,
            status,
            started_at,
            finished_at,
            returncode,
            stdout,
            stderr,
            trigger_source,
            triggered_by_username,
            triggered_by_display_name,
            commit_author,
            pusher
        FROM deployments
        WHERE id = ?
        """,
        (deployment_id,),
    ).fetchone()

    con.close()
    return dict(row) if row else None