appfactory-portal/app/db/users.py

from typing import Any

from app.db.database import get_connection
from app.db.migrations import run_migrations

# Pořadí od nejvyšších práv po nejnižší. "guest" je výchozí role nového účtu –
# nemá přístup nikam, jen vidí výzvu k přidělení práv administrátorem.
ROLES = ("admin", "developer", "viewer", "guest")


def _table_columns(con, table_name: str) -> set[str]:
    rows = con.execute(f"PRAGMA table_info({table_name})").fetchall()
    return {row["name"] for row in rows}


def list_users() -> list[dict[str, Any]]:
    run_migrations()
    con = get_connection()
    columns = _table_columns(con, "users")
    last_login_select = "last_login_at" if "last_login_at" in columns else "NULL AS last_login_at"
    auth_provider_select = "auth_provider" if "auth_provider" in columns else "NULL AS auth_provider"
    provider_subject_select = "provider_subject" if "provider_subject" in columns else "NULL AS provider_subject"
    avatar_url_select = "avatar_url" if "avatar_url" in columns else "NULL AS avatar_url"
    gitea_user_id_select = "gitea_user_id" if "gitea_user_id" in columns else "NULL AS gitea_user_id"
    gitea_username_select = "gitea_username" if "gitea_username" in columns else "NULL AS gitea_username"
    gitea_sync_status_select = "gitea_sync_status" if "gitea_sync_status" in columns else "'not_required' AS gitea_sync_status"
    gitea_sync_error_select = "gitea_sync_error" if "gitea_sync_error" in columns else "NULL AS gitea_sync_error"
    gitea_synced_at_select = "gitea_synced_at" if "gitea_synced_at" in columns else "NULL AS gitea_synced_at"

    rows = con.execute(
        f"""
        SELECT
            id,
            username,
            display_name,
            email,
            role,
            COALESCE(is_enabled, 1) AS is_enabled,
            created_at,
            updated_at,
            {last_login_select},
            {auth_provider_select},
            {provider_subject_select},
            {avatar_url_select},
            {gitea_user_id_select},
            {gitea_username_select},
            {gitea_sync_status_select},
            {gitea_sync_error_select},
            {gitea_synced_at_select}
        FROM users
        ORDER BY id
        """
    ).fetchall()

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


def get_user(user_id: int) -> dict[str, Any] | None:
    run_migrations()
    con = get_connection()
    columns = _table_columns(con, "users")
    last_login_select = "last_login_at" if "last_login_at" in columns else "NULL AS last_login_at"
    auth_provider_select = "auth_provider" if "auth_provider" in columns else "NULL AS auth_provider"
    provider_subject_select = "provider_subject" if "provider_subject" in columns else "NULL AS provider_subject"
    avatar_url_select = "avatar_url" if "avatar_url" in columns else "NULL AS avatar_url"
    gitea_user_id_select = "gitea_user_id" if "gitea_user_id" in columns else "NULL AS gitea_user_id"
    gitea_username_select = "gitea_username" if "gitea_username" in columns else "NULL AS gitea_username"
    gitea_sync_status_select = "gitea_sync_status" if "gitea_sync_status" in columns else "'not_required' AS gitea_sync_status"
    gitea_sync_error_select = "gitea_sync_error" if "gitea_sync_error" in columns else "NULL AS gitea_sync_error"
    gitea_synced_at_select = "gitea_synced_at" if "gitea_synced_at" in columns else "NULL AS gitea_synced_at"

    row = con.execute(
        f"""
        SELECT
            id,
            username,
            display_name,
            email,
            role,
            COALESCE(is_enabled, 1) AS is_enabled,
            created_at,
            updated_at,
            {last_login_select},
            {auth_provider_select},
            {provider_subject_select},
            {avatar_url_select},
            {gitea_user_id_select},
            {gitea_username_select},
            {gitea_sync_status_select},
            {gitea_sync_error_select},
            {gitea_synced_at_select}
        FROM users
        WHERE id = ?
        """,
        (user_id,),
    ).fetchone()

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


def count_enabled_admins(excluding_user_id: int | None = None) -> int:
    run_migrations()
    con = get_connection()
    params: list[Any] = []
    exclusion = ""
    if excluding_user_id is not None:
        exclusion = "AND id != ?"
        params.append(excluding_user_id)

    row = con.execute(
        f"""
        SELECT COUNT(*) AS count
        FROM users
        WHERE LOWER(COALESCE(role, '')) = 'admin'
          AND COALESCE(is_enabled, 1) = 1
          AND COALESCE(is_active, 1) = 1
          {exclusion}
        """,
        params,
    ).fetchone()

    con.close()
    return int(row["count"] or 0)


def update_user_role(user_id: int, role: str) -> None:
    if role not in ROLES:
        raise ValueError("Invalid user role")

    run_migrations()
    con = get_connection()
    con.execute(
        """
        UPDATE users
        SET role = ?,
            updated_at = CURRENT_TIMESTAMP
        WHERE id = ?
        """,
        (role, user_id),
    )
    con.commit()
    con.close()


def set_user_enabled(user_id: int, enabled: bool) -> None:
    run_migrations()
    con = get_connection()
    con.execute(
        """
        UPDATE users
        SET is_enabled = ?,
            updated_at = CURRENT_TIMESTAMP
        WHERE id = ?
        """,
        (1 if enabled else 0, user_id),
    )
    con.commit()
    con.close()


def delete_user(user_id: int) -> None:
    run_migrations()
    con = get_connection()
    con.execute("DELETE FROM users WHERE id = ?", (user_id,))
    con.commit()
    con.close()


def set_gitea_username(user_id: int, gitea_username: str) -> None:
    """Uloží vlastní Gitea uživatelské jméno (handle). Prázdná hodnota = vrácení na automatické portal-{id}."""
    run_migrations()
    con = get_connection()
    con.execute(
        """
        UPDATE users
        SET gitea_username = ?,
            updated_at = CURRENT_TIMESTAMP
        WHERE id = ?
        """,
        ((gitea_username or "").strip() or None, user_id),
    )
    con.commit()
    con.close()


def update_gitea_sync_state(
    user_id: int,
    status: str,
    *,
    gitea_user_id: int | None = None,
    error: str = "",
    synced: bool = False,
) -> None:
    run_migrations()
    con = get_connection()
    if gitea_user_id is None:
        con.execute(
            """
            UPDATE users
            SET gitea_sync_status = ?,
                gitea_sync_error = ?,
                gitea_synced_at = CASE WHEN ? THEN CURRENT_TIMESTAMP ELSE gitea_synced_at END,
                updated_at = CURRENT_TIMESTAMP
            WHERE id = ?
            """,
            (status, error, 1 if synced else 0, user_id),
        )
    else:
        con.execute(
            """
            UPDATE users
            SET gitea_user_id = ?,
                gitea_sync_status = ?,
                gitea_sync_error = ?,
                gitea_synced_at = CASE WHEN ? THEN CURRENT_TIMESTAMP ELSE gitea_synced_at END,
                updated_at = CURRENT_TIMESTAMP
            WHERE id = ?
            """,
            (gitea_user_id, status, error, 1 if synced else 0, user_id),
        )
    con.commit()
    con.close()