devplacepy/devplacepy/database/users.py

# retoor <retoor@molodetz.nl>

from .core import TTLCache, bump_cache_version, db, sync_local_cache


def get_users_by_uids(uids):
    if not uids or "users" not in db.tables:
        return {}
    users = db["users"]
    if "uid" not in users.columns:
        return {}
    seen = set()
    unique = [u for u in uids if u not in seen and not seen.add(u)]
    return {u["uid"]: u for u in users.find(users.table.columns.uid.in_(unique))}


_admins_cache = TTLCache(ttl=300, max_size=4)
# The primary administrator must be an account that can actually authenticate, so scan a
# few of the earliest admins and skip any that are soft-deleted or deactivated.
PRIMARY_ADMIN_CANDIDATES = 50


def invalidate_admins_cache() -> None:
    _admins_cache.clear()
    bump_cache_version("admins")


def get_admin_uids():
    sync_local_cache("admins", _admins_cache)
    cached = _admins_cache.get("uids")
    if cached is not None:
        return list(cached)
    if "users" not in db.tables:
        return []
    rows = db.query("SELECT uid FROM users WHERE role = 'Admin'")
    uids = [row["uid"] for row in rows]
    _admins_cache.set("uids", uids)
    return list(uids)


def set_user_timezone(user_uid: str, tz_name: str) -> None:
    if "users" not in db.tables or not user_uid or not tz_name:
        return
    users = db["users"]
    if not users.has_column("timezone"):
        users.create_column_by_example("timezone", "")
    current = users.find_one(uid=user_uid)
    if current and current.get("timezone") == tz_name:
        return
    users.update({"uid": user_uid, "timezone": tz_name}, ["uid"])


def set_last_seen(user_uid: str, iso: str) -> None:
    if "users" not in db.tables or not user_uid or not iso:
        return
    users = db["users"]
    if not users.has_column("last_seen"):
        users.create_column_by_example("last_seen", "")
    users.update({"uid": user_uid, "last_seen": iso}, ["uid"])


def get_online_users(cutoff_iso: str, limit: int = 30) -> list:
    if "users" not in db.tables:
        return []
    users = db["users"]
    if "last_seen" not in users.columns:
        return []
    return list(
        users.find(
            last_seen={">=": cutoff_iso},
            order_by=["username"],
            _limit=limit,
        )
    )


def _can_hold_primary_admin(row, tracks_active):
    if row.get("deleted_at"):
        return False
    if not tracks_active:
        return True
    is_active = row.get("is_active")
    return is_active is None or bool(is_active)


def get_primary_admin_uid():
    sync_local_cache("admins", _admins_cache)
    cached = _admins_cache.get("primary")
    if cached is not None:
        return cached or None
    if "users" not in db.tables:
        return None
    rows = list(
        db.query(
            "SELECT * FROM users WHERE role = 'Admin' "
            "ORDER BY (created_at IS NULL OR created_at = ''), created_at ASC, id ASC "
            "LIMIT :cap",
            cap=PRIMARY_ADMIN_CANDIDATES,
        )
    )
    tracks_active = "is_active" in db["users"].columns
    primary = next(
        (row["uid"] for row in rows if _can_hold_primary_admin(row, tracks_active)),
        None,
    )
    _admins_cache.set("primary", primary or "")
    return primary


def search_users_by_username(q, *, exclude_uid=None, limit=10):
    if not q or "users" not in db.tables:
        return []
    if exclude_uid is not None:
        rows = db.query(
            "SELECT uid, username FROM users WHERE username LIKE :q AND uid != :me LIMIT :limit",
            q=f"%{q}%",
            me=exclude_uid,
            limit=limit,
        )
    else:
        rows = db.query(
            "SELECT uid, username FROM users WHERE username LIKE :q LIMIT :limit",
            q=f"%{q}%",
            limit=limit,
        )
    return [{"uid": r["uid"], "username": r["username"]} for r in rows]