devplacepy/devplacepy/database/ranking.py

# retoor <retoor@molodetz.nl>

import os

from .core import TTLCache, _in_clause, _now_iso, db, get_table
from .users import get_users_by_uids
from .soft_delete import soft_delete, soft_delete_in


VOTABLE_TARGETS: dict[str, str] = {
    "post": "posts",
    "project": "projects",
    "gist": "gists",
    "comment": "comments",
    "quiz": "quizzes",
}


STAR_TARGETS: set[str] = {"post", "project", "gist", "quiz"}


RANKING_TTL = int(os.environ.get("DEVPLACE_RANKING_TTL", "60"))


_authors_cache = TTLCache(ttl=RANKING_TTL, max_size=200)


_stars_cache = TTLCache(ttl=15, max_size=2000)


def _ranked_authors() -> list:
    cached = _authors_cache.get("ranked")
    if cached is not None:
        return cached
    tables = db.tables
    sources = [
        (target_type, table_name)
        for target_type, table_name in VOTABLE_TARGETS.items()
        if table_name in tables
    ]
    if "votes" not in tables or not sources:
        _authors_cache.set("ranked", [])
        return []
    target_union = " UNION ALL ".join(
        f"SELECT uid, user_uid, '{target_type}' AS target_type FROM {table_name} WHERE deleted_at IS NULL"
        for target_type, table_name in sources
    )
    rows = db.query(
        f"SELECT t.user_uid, SUM(v.value) AS total "
        f"FROM votes v JOIN ({target_union}) t ON v.target_uid = t.uid AND v.target_type = t.target_type "
        f"WHERE v.deleted_at IS NULL "
        f"GROUP BY t.user_uid HAVING SUM(v.value) > 0 ORDER BY total DESC"
    )
    ranked = [(row["user_uid"], row["total"]) for row in rows]
    users_map = get_users_by_uids([uid for uid, _ in ranked])
    authors = []
    for uid, total in ranked:
        user = users_map.get(uid)
        if user:
            author = dict(user)
            author["stars"] = total
            authors.append(author)
    _authors_cache.set("ranked", authors)
    _authors_cache.set(
        "rank_map",
        {author["uid"]: position for position, author in enumerate(authors, start=1)},
    )
    return authors


def _rank_map() -> dict:
    cached = _authors_cache.get("rank_map")
    if cached is not None:
        return cached
    _ranked_authors()
    return _authors_cache.get("rank_map") or {}


def get_top_authors(limit: int = 5) -> list:
    return _ranked_authors()[:limit]


def get_leaderboard(limit: int = 50, offset: int = 0) -> list:
    sliced = _ranked_authors()[offset : offset + limit]
    leaderboard = []
    for position, author in enumerate(sliced, start=offset + 1):
        entry = dict(author)
        entry["rank"] = position
        leaderboard.append(entry)
    return leaderboard


def get_user_rank(user_uid: str):
    return _rank_map().get(user_uid)


def clear_user_stars(user_uid: str) -> None:
    _stars_cache.pop(user_uid)


def get_user_stars(user_uid: str) -> int:
    cached = _stars_cache.get(user_uid)
    if cached is not None:
        return cached
    tables = db.tables
    if "votes" not in tables:
        return 0
    target_union = " UNION ALL ".join(
        f"SELECT uid, '{target_type}' AS target_type FROM {table_name} WHERE user_uid = :u AND deleted_at IS NULL"
        for target_type, table_name in VOTABLE_TARGETS.items()
        if table_name in tables
    )
    if not target_union:
        return 0
    total = 0
    for row in db.query(
        f"SELECT COALESCE(SUM(v.value), 0) AS s "
        f"FROM votes v JOIN ({target_union}) t ON v.target_uid = t.uid AND v.target_type = t.target_type "
        f"WHERE v.deleted_at IS NULL",
        u=user_uid,
    ):
        total = row["s"] or 0
        break
    _stars_cache.set(user_uid, total)
    return total


def update_target_stars(target_type: str, target_uid: str, net_stars: int) -> None:
    table_name = VOTABLE_TARGETS.get(target_type)
    if not table_name:
        return
    if target_type in STAR_TARGETS:
        get_table(table_name).update({"uid": target_uid, "stars": net_stars}, ["uid"])


def soft_delete_engagement(target_type: str, target_uids: list, deleted_by: str) -> None:
    uids = [uid for uid in (target_uids or []) if uid]
    if not uids:
        return
    stamp = _now_iso()
    soft_delete_in(
        "reactions", "target_uid", uids, deleted_by, stamp=stamp, target_type=target_type
    )
    soft_delete_in(
        "bookmarks", "target_uid", uids, deleted_by, stamp=stamp, target_type=target_type
    )
    if target_type == "post" and "polls" in db.tables:
        for uid in uids:
            for poll in db["polls"].find(post_uid=uid, deleted_at=None):
                soft_delete("poll_votes", deleted_by, stamp=stamp, poll_uid=poll["uid"])
                soft_delete("poll_options", deleted_by, stamp=stamp, poll_uid=poll["uid"])
            soft_delete("polls", deleted_by, stamp=stamp, post_uid=uid)


def delete_engagement(target_type: str, target_uids: list) -> None:
    uids = [uid for uid in (target_uids or []) if uid]
    if not uids:
        return
    tables = db.tables
    if "reactions" in tables:
        placeholders, params = _in_clause(uids)
        params["tt"] = target_type
        with db:
            db.query(
                f"DELETE FROM reactions WHERE target_type=:tt AND target_uid IN ({placeholders})",
                **params,
            )
    if "bookmarks" in tables:
        placeholders, params = _in_clause(uids)
        params["tt"] = target_type
        with db:
            db.query(
                f"DELETE FROM bookmarks WHERE target_type=:tt AND target_uid IN ({placeholders})",
                **params,
            )
    if target_type == "post" and "polls" in tables:
        for uid in uids:
            for poll in db["polls"].find(post_uid=uid):
                if "poll_votes" in tables:
                    db["poll_votes"].delete(poll_uid=poll["uid"])
                if "poll_options" in tables:
                    db["poll_options"].delete(poll_uid=poll["uid"])
            db["polls"].delete(post_uid=uid)


def get_target_owner_uid(target_type: str, target_uid: str) -> str | None:
    table_name = VOTABLE_TARGETS.get(target_type)
    if not table_name:
        return None
    row = get_table(table_name).find_one(uid=target_uid, deleted_at=None)
    return row["user_uid"] if row else None