#FastAPI#ProductionAdvanced

Gamification System: Build XP, Badges & Leaderboards

Jun 7, 2026
18 min read

AI Insights

Powered by GPT-4o-mini

Verified Context: gamification-system-build-xp-badges-leaderboards
Quick Answer

Event-driven XP system, badge criteria engine, streak tracking, leaderboard ranking, milestone notifications. Database schema, trigger logic, and the math behind leveling.

Quick Summary

Learn to build a custom gamification system with XP, badges, and leaderboards using FastAPI and PostgreSQL. Enhance user engagement today!

Gamification: Building XP, Badges, and Leaderboards From Scratch

This blog has a gamification system: users earn XP for reading posts, unlock badges for milestones, maintain streaks for daily activity, and climb a global leaderboard. No third-party service — it's all custom-built in PostgreSQL and FastAPI.

The Event-Driven XP System

Every user action that should award XP emits an event:

python
class XPEvent(Enum):
    POST_READ = ("post_read", 10)
    POST_COMPLETED = ("post_completed", 25)
    SERIES_COMPLETED = ("series_completed", 100)
    CHALLENGE_SOLVED = ("challenge_solved", 30)
    COMMENT_LEFT = ("comment_left", 5)
    STREAK_MILESTONE = ("streak_milestone", 50)
    BADGE_EARNED = ("badge_earned", 0)  # No XP, badge is the reward

    def __init__(self, event_type: str, base_xp: int):
        self.event_type = event_type
        self.base_xp = base_xp

Events are processed asynchronously through a Redis pub/sub channel:

python
async def award_xp(user_id: str, event: XPEvent, multiplier: float = 1.0):
    xp = int(event.base_xp * multiplier)
    await redis.publish("xp_events", json.dumps({
        "user_id": user_id,
        "xp": xp,
        "event_type": event.event_type,
        "timestamp": datetime.utcnow().isoformat(),
    }))

A background worker consumes the channel and writes to the database:

python
async def process_xp_events():
    pubsub = redis.pubsub()
    await pubsub.subscribe("xp_events")

    async for message in pubsub.listen():
        if message["type"] != "message":
            continue

        data = json.loads(message["data"])
        async with db.begin():
            user = await db.get(User, data["user_id"])
            user.xp += data["xp"]
            user.updated_at = func.now()

            await check_level_up(user, db)
            await check_badges(user, db, data["event_type"])
            await update_leaderboard(user)

The Leveling Curve

XP maps to levels using a quadratic formula:

python
def xp_for_level(level: int) -> int:
    return int(100 * (level ** 1.5))

Level 1: 100 XP. Level 10: 3162 XP. Level 20: 8944 XP. Level 50: 35355 XP.

The square-root relationship means early levels come quickly (good for engagement) and later levels take progressively more effort (good for retention).

python
def level_from_xp(total_xp: int) -> int:
    return int((total_xp / 100) ** (2 / 3))

When a user levels up, a notification is created:

python
async def check_level_up(user, db):
    new_level = level_from_xp(user.xp)
    if new_level > user.level:
        old_level = user.level
        user.level = new_level
        db.add(Notification(
            user_id=user.id,
            type="level_up",
            title=f"Level {new_level}!",
            message=f"You advanced from level {old_level} to {new_level}.",
        ))

The Badge Criteria Engine

Badges are defined in the database with a criteria JSONB field:

python
class Badge(Base):
    __tablename__ = "badges"

    id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
    name: Mapped[str]
    description: Mapped[str]
    icon: Mapped[str]
    criteria: Mapped[dict] = mapped_column(JSONB)

The criteria field contains a JSON object that defines the condition:

json
{
    "type": "posts_read",
    "count": 10,
    "scope": "total"
}

{
    "type": "streak",
    "count": 7,
    "scope": "consecutive_days"
}

{
    "type": "series_complete",
    "count": 3,
    "scope": "unique_series"
}

{
    "type": "xp_total",
    "count": 5000,
    "scope": "lifetime"
}

{
    "type": "challenge_streak",
    "count": 5,
    "scope": "consecutive_days"
}

The badge checker runs after every XP event:

python
async def check_badges(user, db, event_type: str):
    badges = await db.execute(select(Badge))
    user_badge_ids = {ub.badge_id for ub in user.badges}

    for badge in badges.scalars():
        if badge.id in user_badge_ids:
            continue

        if await evaluate_criteria(user, badge.criteria, event_type, db):
            user.badges.append(badge)
            db.add(Notification(
                user_id=user.id,
                type="badge_earned",
                title=f"Badge Unlocked: {badge.name}",
                message=badge.description,
            ))
            await award_xp(user.id, XPEvent.BADGE_EARNED)

The evaluate_criteria function interprets the badge's criteria JSON:

python
async def evaluate_criteria(user, criteria, event_type, db) -> bool:
    if criteria["type"] == "posts_read":
        result = await db.execute(
            select(func.count(UserProgress.id)).where(
                UserProgress.user_id == user.id,
            )
        )
        return result.scalar() >= criteria["count"]

    if criteria["type"] == "streak":
        return user.streak >= criteria["count"]

    if criteria["type"] == "xp_total":
        return user.xp >= criteria["count"]

    return False

Streak Tracking

The streak is computed daily from the user_progress table:

python
async def update_streak(user_id: str, db):
    today = date.today()

    last_activity = await db.execute(
        select(func.max(UserProgress.completed_at)).where(
            UserProgress.user_id == user_id,
        )
    )
    last_date = last_activity.scalar()

    if last_date and last_date.date() == today - timedelta(days=1):
        user.streak += 1
        if user.streak % 7 == 0:
            await award_xp(user.id, XPEvent.STREAK_MILESTONE)
    elif last_date and last_date.date() < today - timedelta(days=1):
        user.streak = 0

A midnight cron job checks for streaks that are about to expire and sends a push notification (if the user has enabled them): "Your 14-day streak expires tomorrow — read one post to keep it alive!"

Leaderboard

The leaderboard is stored as a Redis sorted set:

python
async def update_leaderboard(user):
    await redis.zadd("leaderboard", {user.id: user.xp})

Reading the top N:

python
async def get_leaderboard(limit: int = 100, offset: int = 0):
    results = await redis.zrevrange("leaderboard", offset, offset + limit - 1, withscores=True)
    return [
        {"user_id": user_id.decode(), "xp": int(xp)}
        for user_id, xp in results
    ]

The frontend polls every 30 seconds:

typescript
const { data } = useQuery({
    queryKey: ["leaderboard"],
    queryFn: () => fetch("/api/v1/gamification/leaderboard").then((r) => r.json()),
    refetchInterval: 30000,
});

A materialized PostgreSQL view recomputes the leaderboard every 5 seconds as a fallback:

sql
CREATE MATERIALIZED VIEW leaderboard_mv AS
SELECT id, username, xp, level,
       RANK() OVER (ORDER BY xp DESC) AS rank
FROM users
WHERE xp > 0;

The Redis sorted set is the primary read layer; the materialized view is the source of truth for consistency checks.

Frontend Components

BadgeGrid

Displays earned badges with a lock/unlock animation:

tsx
function BadgeGrid({ badges, earnedBadgeIds }: Props) {
    return (
        <div className="grid grid-cols-3 sm:grid-cols-4 md:grid-cols-6 gap-4">
            {badges.map((badge) => (
                <div
                    key={badge.id}
                    className={`card p-4 text-center transition-all duration-300 ${
                        earnedBadgeIds.has(badge.id)
                            ? "border-amber-500/30 bg-amber-500/5"
                            : "opacity-40 grayscale"
                    }`}
                >
                    <span className="text-3xl">{badge.icon}</span>
                    <p className="text-[10px] font-bold mt-2 uppercase tracking-wider">
                        {badge.name}
                    </p>
                </div>
            ))}
        </div>
    );
}

UserStats

Shows XP, level, streak, and rank in a compact card:

tsx
function UserStats({ stats }: { stats: UserStats }) {
    const xpProgress = (stats.xp % xpForLevel(stats.level)) / xpForLevel(stats.level + 1) * 100;

    return (
        <div className="card p-5 space-y-4">
            <div className="flex justify-between items-center">
                <span className="text-[10px] font-bold uppercase tracking-widest text-white/40">
                    Level {stats.level}
                </span>
                <span className="text-sm font-bold text-amber-500">
                    #{stats.rank}
                </span>
            </div>
            <div className="w-full h-2 rounded-full bg-white/[0.06] overflow-hidden">
                <div
                    className="h-full rounded-full bg-gradient-to-r from-amber-500 to-amber-400 transition-all duration-500"
                    style={{ width: `${xpProgress}%` }}
                />
            </div>
            <div className="flex justify-between text-xs text-white/50">
                <span>{stats.xp.toLocaleString()} XP</span>
                <span>{stats.streak} day streak</span>
            </div>
        </div>
    );
}

What's Next

The next post covers the Spaced Repetition System — how the SM-2 algorithm is implemented, how review schedules are computed, and how reading progress feeds into the retention model.

Built with FastAPI, PostgreSQL, Redis sorted sets, and zero third-party gamification services.