Motor de gamification: Arhitectură și mașină de stat
80% dintre studenți spun că sunt mai motivați atunci când studiază pe platforme care încorporează elemente de joc. Piața de gamification în EdTech va crește la a CAGR peste 36% până în 2032. Dar diferența dintre o gamificare eficient și superficial nu este despre adăugarea de insigne și puncte aleatorii: este vorba de soliditate alarhitectura de baza.
Un motor de gamification prost conceput acumulează rapid datoria tehnică: logica progresie codificată, starea cursei la punctele când mulți utilizatori completează activități în același timp, clasamente care devin blocaje pentru mii de utilizatori, insigne acordate de mai multe ori din cauza erorilor în gestionarea stării. Aceste probleme nu sunt teoretice: sunt patologiile comune ale fiecărei platforme care adaugă gamification ca o idee ulterioară în loc să-l proiecteze ca o caracteristică de primă clasă.
În acest articol vom construi un Motor de gamification robust și scalabil cu mașini de stare explicite, sursă de evenimente pentru trasabilitate, Redis pentru clasamente în timp real și un sistem de reguli configurabil care permite creatorilor de conținut să definească mecanică nouă fără a atinge codul.
Ce veți învăța în acest articol
- Arhitectură bazată pe evenimente a unui motor de gamification scalabil
- Mașină de stat pentru gestionarea progresiei elevilor
- Sistem de insigne cu reguli configurabile (fără hardcoding)
- Clasament în timp real cu seturi sortate Redis
- Sistem de urmărire și recompensă în timp
- XP (puncte de experiență) cu niveluri și curbe de progresie
- Anti-modele: cum să evitați condițiile de cursă și dubla atribuire
- Notificări push pentru realizări cu WebSocket
1. Principii de proiectare: Gamificare intrinsecă vs extrinsecă
Înainte de a scrie cod, este esențial să înțelegeți diferența dintre motivație extrinseci (puncte, insigne, clasamente - motivatori externi) e intrinsec (curiozitate, simț al competenței, autonomie). Gamificarea educațională de succes folosește motivatori extrinseci ca rampă de lansare pentru a dezvolta motivația intrinsecă, nu ca substitut.
Cele mai eficiente mecanisme de gamification pentru educație în 2025 sunt: dâră (secvențe de zile consecutive de studiu), progresia stăpânirii (vizualizarea progresului către stăpânire), învăţarea socială (clasare în rândul colegilor cu niveluri similare), e alegeri semnificative (căi alternative cu recompense diferite). Cele mai puțin eficiente (și potențial dăunătoare) mecanice sunt clasamentele globale unde noii studenți văd întotdeauna pe cei mai performanti ca fiind imposibil de atins.
Componentele motorului de gamification
| Componentă | Responsabilitate | Tehnologie |
|---|---|---|
| Autobuz de evenimente | Colectează evenimente de învățare | Fluxuri Kafka/Redis |
| Rule Engine | Evaluați regulile configurabile | Reguli Python + JSON |
| Mașină de stat | Gestionează progresul elevilor | Python FSM personalizat |
| Calculator XP | Calculați punctele de experiență | Python + PostgreSQL |
| Manager de insigne | Premiu insigna idempotent | Cache PostgreSQL + Redis |
| Clasamente | Clasamente în timp real | Seturi sortate Redis |
| Urmăritor de linii | Monitorizați secvențele de studiu | Redis + PostgreSQL |
| Centru de notificare | Împingeți realizarea în timp real | WebSocket / SSE |
2. Mașină de stat pentru progresul elevilor
Mașina de stat este inima motorului de gamification. Fiecare elev este într-unul stat care reflectă nivelul său de implicare și progres. Tranzițiile între state au loc ca răspuns la evenimente (finalizare lecție, răspuns corect, autentificare consecutivă). Faceți această mașină de stare explicită elimină problema „stărilor implicite” care provoacă erori greu de reprodus.
# gamification/state_machine.py
from enum import Enum, auto
from dataclasses import dataclass, field
from typing import Dict, List, Callable, Optional
from datetime import datetime
class LearnerState(Enum):
"""Stati possibili di un learner nella piattaforma."""
ONBOARDING = auto() # Primo accesso, < 3 corsi completati
ACTIVE = auto() # Studente attivo, streak in corso
AT_RISK = auto() # Nessuna attivita da 7-13 giorni
DORMANT = auto() # Inattivo da 14+ giorni
MASTERY = auto() # Ha completato un corso con punteggio >= 90%
MENTOR = auto() # Ha aiutato almeno 5 altri studenti
class LearnerEvent(Enum):
"""Eventi che causano transizioni di stato."""
LESSON_COMPLETED = "lesson_completed"
QUIZ_PASSED = "quiz_passed"
QUIZ_FAILED = "quiz_failed"
COURSE_COMPLETED = "course_completed"
PEER_HELPED = "peer_helped"
DAILY_LOGIN = "daily_login"
STREAK_BROKEN = "streak_broken"
INACTIVE_7_DAYS = "inactive_7_days"
INACTIVE_14_DAYS = "inactive_14_days"
REACTIVATED = "reactivated"
@dataclass
class StateTransition:
from_state: LearnerState
event: LearnerEvent
to_state: LearnerState
conditions: List[Callable] = field(default_factory=list)
actions: List[Callable] = field(default_factory=list)
@dataclass
class LearnerContext:
student_id: str
current_state: LearnerState
xp: int = 0
streak_days: int = 0
courses_completed: int = 0
peers_helped: int = 0
last_activity: Optional[datetime] = None
metadata: Dict = field(default_factory=dict)
class LearnerStateMachine:
"""
State machine per la progressione del learner.
Tutte le transizioni sono esplicite e testabili.
"""
TRANSITIONS: List[StateTransition] = [
# ONBOARDING -> ACTIVE: ha completato la prima lezione
StateTransition(
from_state=LearnerState.ONBOARDING,
event=LearnerEvent.LESSON_COMPLETED,
to_state=LearnerState.ACTIVE,
),
# ACTIVE -> AT_RISK: 7 giorni di inattivita
StateTransition(
from_state=LearnerState.ACTIVE,
event=LearnerEvent.INACTIVE_7_DAYS,
to_state=LearnerState.AT_RISK,
),
# AT_RISK -> DORMANT: altri 7 giorni senza attivita
StateTransition(
from_state=LearnerState.AT_RISK,
event=LearnerEvent.INACTIVE_14_DAYS,
to_state=LearnerState.DORMANT,
),
# DORMANT/AT_RISK -> ACTIVE: torna ad essere attivo
StateTransition(
from_state=LearnerState.AT_RISK,
event=LearnerEvent.REACTIVATED,
to_state=LearnerState.ACTIVE,
),
StateTransition(
from_state=LearnerState.DORMANT,
event=LearnerEvent.REACTIVATED,
to_state=LearnerState.ACTIVE,
),
# ACTIVE -> MASTERY: completa corso con voto alto
StateTransition(
from_state=LearnerState.ACTIVE,
event=LearnerEvent.COURSE_COMPLETED,
to_state=LearnerState.MASTERY,
conditions=[lambda ctx: ctx.metadata.get("score", 0) >= 90],
),
# MASTERY -> MENTOR: aiuta abbastanza peer
StateTransition(
from_state=LearnerState.MASTERY,
event=LearnerEvent.PEER_HELPED,
to_state=LearnerState.MENTOR,
conditions=[lambda ctx: ctx.peers_helped >= 5],
),
]
def __init__(self, event_publisher):
self.event_publisher = event_publisher
# Indice per lookup rapido O(1)
self._transition_index: Dict[tuple, StateTransition] = {
(t.from_state, t.event): t
for t in self.TRANSITIONS
}
def process_event(
self,
context: LearnerContext,
event: LearnerEvent,
event_data: Dict = None,
) -> LearnerContext:
"""
Processa un evento e restituisce un NUOVO context (immutabile).
Non modifica mai il context in input.
"""
event_data = event_data or {}
key = (context.current_state, event)
transition = self._transition_index.get(key)
if not transition:
# Nessuna transizione definita: evento ignorato
return context
# Aggiorna il metadata del context con i dati dell'evento
updated_metadata = {**context.metadata, **event_data}
temp_context = LearnerContext(
**{
**context.__dict__,
"metadata": updated_metadata,
}
)
# Verifica tutte le condizioni
if not all(cond(temp_context) for cond in transition.conditions):
return context # Condizioni non soddisfatte, nessuna transizione
# Crea nuovo context con nuovo stato
new_context = LearnerContext(
**{
**temp_context.__dict__,
"current_state": transition.to_state,
"last_activity": datetime.utcnow(),
}
)
# Pubblica evento di transizione
self.event_publisher.publish({
"type": "state_transition",
"student_id": context.student_id,
"from_state": context.current_state.name,
"to_state": transition.to_state.name,
"trigger_event": event.value,
"timestamp": datetime.utcnow().isoformat(),
})
return new_context3. XP și sistem de niveluri cu curbe de progresie
Sistemul de puncte de experiență (XP) trebuie să echilibreze viteza de progres: prea repede și studenții ating maxim în câteva zile și se plictisesc, prea încet și se descurajează. Să folosim unul curba logaritmică pentru primele niveluri (progresie rapidă și satisfăcătoare) devine mai abruptă spre niveluri înalte (obiective semnificative).
# gamification/xp_system.py
import math
from dataclasses import dataclass
from typing import Dict, List
from enum import Enum
class ActivityType(Enum):
LESSON_COMPLETED = "lesson"
QUIZ_PASSED = "quiz_pass"
QUIZ_PERFECT = "quiz_perfect" # 100% corretto
COURSE_COMPLETED = "course"
STREAK_BONUS = "streak"
PEER_REVIEW = "peer_review"
HELP_PEER = "help_peer"
FIRST_LOGIN_DAY = "first_login"
# XP base per ogni tipo di attivita
XP_BASE_REWARDS: Dict[str, int] = {
ActivityType.LESSON_COMPLETED.value: 50,
ActivityType.QUIZ_PASSED.value: 100,
ActivityType.QUIZ_PERFECT.value: 200,
ActivityType.COURSE_COMPLETED.value: 500,
ActivityType.STREAK_BONUS.value: 25,
ActivityType.PEER_REVIEW.value: 75,
ActivityType.HELP_PEER.value: 150,
ActivityType.FIRST_LOGIN_DAY.value: 10,
}
@dataclass
class XPGrant:
student_id: str
activity_type: str
base_xp: int
multiplier: float
final_xp: int
new_total_xp: int
new_level: int
level_up: bool
reason: str
class XPCalculator:
"""
Calcola XP con moltiplicatori basati su streak, difficolta e tempo.
Usa la formula logaritmica per i livelli.
"""
def calculate_level(self, total_xp: int) -> int:
"""
Livello = floor(log2(total_xp / 100)) + 1
Esempi: 100 XP = Lv1, 200 XP = Lv2, 400 XP = Lv3, 800 XP = Lv4...
"""
if total_xp < 100:
return 1
return min(int(math.log2(total_xp / 100)) + 1, 100)
def xp_for_level(self, level: int) -> int:
"""XP necessari per raggiungere il livello."""
if level <= 1:
return 0
return 100 * (2 ** (level - 1))
def calculate_multiplier(
self,
streak_days: int,
difficulty: int, # 1-5
time_of_day_bonus: bool = False,
) -> float:
"""
Calcola il moltiplicatore XP.
- Streak: +5% per ogni giorno consecutivo (max +50%)
- Difficolta: +10% per ogni livello sopra il baseline (2)
- Bonus orario: +20% per studio nelle ore ottimali (8-12, 14-18)
"""
streak_bonus = min(streak_days * 0.05, 0.5) # Max 50%
difficulty_bonus = max(0, (difficulty - 2) * 0.10) # Baseline difficolta = 2
time_bonus = 0.20 if time_of_day_bonus else 0.0
return 1.0 + streak_bonus + difficulty_bonus + time_bonus
def grant_xp(
self,
student_id: str,
activity: ActivityType,
current_xp: int,
streak_days: int,
difficulty: int = 2,
time_of_day_bonus: bool = False,
) -> XPGrant:
base_xp = XP_BASE_REWARDS.get(activity.value, 10)
multiplier = self.calculate_multiplier(streak_days, difficulty, time_of_day_bonus)
final_xp = int(base_xp * multiplier)
new_total = current_xp + final_xp
old_level = self.calculate_level(current_xp)
new_level = self.calculate_level(new_total)
return XPGrant(
student_id=student_id,
activity_type=activity.value,
base_xp=base_xp,
multiplier=multiplier,
final_xp=final_xp,
new_total_xp=new_total,
new_level=new_level,
level_up=new_level > old_level,
reason=f"{activity.value} x{multiplier:.2f} (streak:{streak_days}d, diff:{difficulty}/5)",
)4. Sistem de insigne cu Idempotency și Reguli configurabile
Cea mai frecventă problemă în sistemele de insigne este atribuirea dublă: aceeași insignă este acordat de mai multe ori pentru condițiile de cursă sau reîncercări ale evenimentelor eșuate. Soluția șiidempotenta: fiecare operațiune de atribuire a insignei poate fi efectuată de mai multe ori, producând întotdeauna același rezultat. Folosim a constrângere UNIQUE în PostgreSQL ca asigurare la nivel de bază de date.
# gamification/badge_manager.py
import json
from dataclasses import dataclass
from typing import List, Dict, Any, Optional
from datetime import datetime
from enum import Enum
class BadgeTier(Enum):
BRONZE = "bronze"
SILVER = "silver"
GOLD = "gold"
PLATINUM = "platinum"
LEGENDARY = "legendary"
@dataclass
class BadgeDefinition:
badge_id: str
name: str
description: str
tier: BadgeTier
icon_url: str
# Regola come JSON (no hardcoding!)
rule: Dict[str, Any]
# Esempi di regole configurabili
BADGE_CATALOG: List[BadgeDefinition] = [
BadgeDefinition(
badge_id="first_lesson",
name="Primo Passo",
description="Hai completato la tua prima lezione",
tier=BadgeTier.BRONZE,
icon_url="/badges/first-lesson.svg",
rule={"event": "lesson_completed", "count": 1},
),
BadgeDefinition(
badge_id="week_streak",
name="Settimana di Fuoco",
description="7 giorni di studio consecutivi",
tier=BadgeTier.SILVER,
icon_url="/badges/week-streak.svg",
rule={"metric": "streak_days", "gte": 7},
),
BadgeDefinition(
badge_id="perfect_quiz",
name="Perfezionista",
description="Quiz completato con 100% di risposte corrette",
tier=BadgeTier.GOLD,
icon_url="/badges/perfect-quiz.svg",
rule={"event": "quiz_completed", "score": 100},
),
BadgeDefinition(
badge_id="mentor",
name="Mentore della Comunita",
description="Hai aiutato 10 compagni di corso",
tier=BadgeTier.PLATINUM,
icon_url="/badges/mentor.svg",
rule={"metric": "peers_helped", "gte": 10},
),
]
class BadgeManager:
def __init__(self, db, redis_client):
self.db = db
self.redis = redis_client
self._catalog: Dict[str, BadgeDefinition] = {b.badge_id: b for b in BADGE_CATALOG}
async def evaluate_and_award(
self,
student_id: str,
event: str,
student_metrics: Dict[str, Any],
) -> List[str]:
"""
Valuta tutti i badge applicabili e li assegna (idempotente).
Ritorna la lista di badge_id appena assegnati.
"""
newly_awarded = []
for badge_id, badge_def in self._catalog.items():
if await self._is_already_awarded(student_id, badge_id):
continue
if self._evaluate_rule(badge_def.rule, event, student_metrics):
awarded = await self._award_badge_idempotent(student_id, badge_id)
if awarded:
newly_awarded.append(badge_id)
return newly_awarded
async def _award_badge_idempotent(self, student_id: str, badge_id: str) -> bool:
"""
Assegna il badge usando INSERT ... ON CONFLICT DO NOTHING.
Garantisce idempotenza a livello DB.
Ritorna True se il badge e stato effettivamente assegnato ora.
"""
result = await self.db.execute(
"""INSERT INTO student_badges (student_id, badge_id, awarded_at)
VALUES (:sid, :bid, :ts)
ON CONFLICT (student_id, badge_id) DO NOTHING
RETURNING badge_id""",
{"sid": student_id, "bid": badge_id, "ts": datetime.utcnow()},
)
rows = result.fetchall()
if rows:
# Invalida cache badge studente
await self.redis.delete(f"student:badges:{student_id}")
await self.db.commit()
return True
return False # Già esisteva
async def _is_already_awarded(self, student_id: str, badge_id: str) -> bool:
"""Controlla cache Redis prima di interrogare il DB."""
cache_key = f"student:badges:{student_id}"
cached = await self.redis.get(cache_key)
if cached:
badges = json.loads(cached)
return badge_id in badges
# Cache miss: interroga DB
result = await self.db.execute(
"SELECT badge_id FROM student_badges WHERE student_id = :sid",
{"sid": student_id},
)
badges = [row[0] for row in result.fetchall()]
await self.redis.setex(cache_key, 300, json.dumps(badges)) # TTL 5 minuti
return badge_id in badges
def _evaluate_rule(
self,
rule: Dict,
current_event: str,
metrics: Dict[str, Any],
) -> bool:
"""Valuta una regola badge configurabile."""
# Regola basata su evento
if "event" in rule:
if rule["event"] != current_event:
return False
if "count" in rule:
return metrics.get(f"event_count_{rule['event']}", 0) >= rule["count"]
if "score" in rule:
return metrics.get("last_score", 0) >= rule["score"]
return True
# Regola basata su metrica
if "metric" in rule:
metric_val = metrics.get(rule["metric"], 0)
if "gte" in rule:
return metric_val >= rule["gte"]
if "eq" in rule:
return metric_val == rule["eq"]
if "lte" in rule:
return metric_val <= rule["lte"]
return False5. Clasament în timp real cu seturi sortate Redis
Clasamentele globale sunt o sabie cu două tăișuri: îi motivează pe cei mai buni performanți dar îi demotivează pe cei de la capătul clasamentului. Soluția modernă este clasament segmentat: fiecare elev concurează cu 50 de utilizatori mai aproape de scorul său, făcând competiția întotdeauna semnificativă. Seturile Sortate Redis cu complexitate O(log N) pentru toate operațiunile sunt instrumentul perfect pentru clasamente în timp real.
# gamification/leaderboard.py
import json
from typing import List, Dict, Optional
from datetime import datetime, date
import redis.asyncio as redis
@dataclass
class LeaderboardEntry:
rank: int
student_id: str
display_name: str
xp: int
level: int
avatar_url: str
is_current_user: bool = False
class LeaderboardManager:
"""
Leaderboard multi-scope con Redis Sorted Sets.
Scope: global, course-specific, weekly, peer-group.
"""
def __init__(self, redis_client: redis.Redis):
self.redis = redis_client
def _key(self, scope: str, period: str = "all-time") -> str:
"""Genera chiave Redis per uno scope e periodo."""
if period == "weekly":
week = date.today().isocalendar()[1]
year = date.today().year
return f"leaderboard:{scope}:{year}:w{week}"
return f"leaderboard:{scope}:{period}"
async def update_score(
self,
student_id: str,
xp_delta: int,
scopes: List[str],
) -> None:
"""
Aggiorna il punteggio in tutti gli scope rilevanti.
Usa pipeline Redis per operazioni atomiche.
"""
async with self.redis.pipeline(transaction=True) as pipe:
for scope in scopes:
for period in ["all-time", "weekly"]:
key = self._key(scope, period)
pipe.zincrby(key, xp_delta, student_id)
# Imposta TTL per leaderboard settimanale
if period == "weekly":
pipe.expire(key, 7 * 24 * 3600)
await pipe.execute()
async def get_leaderboard(
self,
scope: str,
page: int = 1,
page_size: int = 10,
period: str = "all-time",
) -> List[Dict]:
"""Recupera la leaderboard paginata per uno scope."""
key = self._key(scope, period)
start = (page - 1) * page_size
end = start + page_size - 1
# ZREVRANGE: ordine decrescente (punteggio più alto prima)
entries = await self.redis.zrevrange(key, start, end, withscores=True)
result = []
for rank_offset, (student_id_bytes, score) in enumerate(entries):
student_id = student_id_bytes.decode() if isinstance(student_id_bytes, bytes) else student_id_bytes
result.append({
"rank": start + rank_offset + 1,
"student_id": student_id,
"xp": int(score),
})
return result
async def get_peer_leaderboard(
self,
student_id: str,
scope: str,
window: int = 25,
period: str = "all-time",
) -> List[Dict]:
"""
Leaderboard contestualizzata: mostra i 25 utenti sopra e sotto il corrente studente.
"""
key = self._key(scope, period)
# Posizione attuale dello studente (rank 0-indexed dal basso)
rank_from_bottom = await self.redis.zrank(key, student_id)
if rank_from_bottom is None:
return []
total = await self.redis.zcard(key)
rank_from_top = total - rank_from_bottom - 1
# Finestra: window utenti sopra e sotto
start = max(0, rank_from_top - window)
end = min(total - 1, rank_from_top + window)
entries = await self.redis.zrevrange(key, start, end, withscores=True)
result = []
for i, (sid_bytes, score) in enumerate(entries):
sid = sid_bytes.decode() if isinstance(sid_bytes, bytes) else sid_bytes
result.append({
"rank": start + i + 1,
"student_id": sid,
"xp": int(score),
"is_current_user": sid == student_id,
})
return result
async def get_student_rank(
self,
student_id: str,
scope: str,
period: str = "all-time",
) -> Optional[int]:
"""Recupera il rank assoluto di uno studente."""
key = self._key(scope, period)
total = await self.redis.zcard(key)
rank_from_bottom = await self.redis.zrank(key, student_id)
if rank_from_bottom is None:
return None
return total - rank_from_bottom # Rank dal top (1 = primo posto)6. Urmărirea seriei și recompense temporale
Streak-urile sunt una dintre cele mai puternice mecanisme de implicare din EdTech. Duolingo și-a construit un imperiu pe conceptul de „nu rupe stria”. Implementarea trebuie să gestioneze corect fusurile orare (un student în Japonia nu trebuie să-și piardă seria pentru că a studiat la 23:59, ora locală) și „îngheață” (zile de grație pentru evenimente neașteptate).
# gamification/streak_tracker.py
from datetime import datetime, date, timedelta
from typing import Optional, Dict
import zoneinfo
import redis.asyncio as redis
@dataclass
class StreakStatus:
student_id: str
current_streak: int
longest_streak: int
last_activity_date: Optional[date]
freeze_remaining: int # Giorni di grazia disponibili
streak_at_risk: bool # True se non ha ancora fatto attivita oggi
class StreakTracker:
FREEZE_MAX = 2 # Massimo 2 giorni di grazia
def __init__(self, redis_client: redis.Redis, db):
self.redis = redis_client
self.db = db
def _today_for_user(self, timezone: str) -> date:
"""Data corrente nel fuso orario dello studente."""
tz = zoneinfo.ZoneInfo(timezone)
return datetime.now(tz).date()
async def record_activity(
self,
student_id: str,
timezone: str = "UTC",
) -> Dict:
"""
Registra attivita dello studente e aggiorna la streak.
Ritorna le modifiche alla streak.
"""
today = self._today_for_user(timezone)
status = await self.get_status(student_id, timezone)
if not status.last_activity_date:
# Prima attivita in assoluto
new_streak = 1
streak_extended = True
elif status.last_activity_date == today:
# Già registrata oggi: nessun cambio
return {"streak_changed": False, "current_streak": status.current_streak}
elif status.last_activity_date == today - timedelta(days=1):
# Ieri: estendi streak
new_streak = status.current_streak + 1
streak_extended = True
elif (today - status.last_activity_date).days <= 1 + status.freeze_remaining:
# Streak salvata da un freeze
days_missed = (today - status.last_activity_date).days - 1
new_freeze = max(0, status.freeze_remaining - days_missed)
new_streak = status.current_streak + 1
streak_extended = True
await self._update_freeze(student_id, new_freeze)
else:
# Streak rotta
new_streak = 1
streak_extended = False
# Salva aggiornamento
await self._save_streak(student_id, new_streak, status.longest_streak, today)
return {
"streak_changed": True,
"streak_extended": streak_extended,
"current_streak": new_streak,
"longest_streak": max(new_streak, status.longest_streak),
"streak_broken": not streak_extended and status.current_streak > 1,
"previous_streak": status.current_streak if not streak_extended else None,
}
async def get_status(self, student_id: str, timezone: str = "UTC") -> StreakStatus:
"""Recupera lo stato streak corrente dalla cache Redis."""
cache_key = f"streak:{student_id}"
cached = await self.redis.hgetall(cache_key)
if cached:
last_date_str = cached.get(b"last_date", b"").decode()
last_date = date.fromisoformat(last_date_str) if last_date_str else None
today = self._today_for_user(timezone)
return StreakStatus(
student_id=student_id,
current_streak=int(cached.get(b"current", b"0").decode()),
longest_streak=int(cached.get(b"longest", b"0").decode()),
last_activity_date=last_date,
freeze_remaining=int(cached.get(b"freeze", b"0").decode()),
streak_at_risk=last_date != today if last_date else True,
)
# Cache miss: leggi dal DB
row = await self.db.execute(
"""SELECT current_streak, longest_streak, last_activity_date, freeze_remaining
FROM student_streaks WHERE student_id = :sid""",
{"sid": student_id},
)
data = row.fetchone()
if not data:
return StreakStatus(student_id, 0, 0, None, self.FREEZE_MAX, True)
today = self._today_for_user(timezone)
return StreakStatus(
student_id=student_id,
current_streak=data[0],
longest_streak=data[1],
last_activity_date=data[2],
freeze_remaining=data[3],
streak_at_risk=data[2] != today if data[2] else True,
)
async def _save_streak(
self,
student_id: str,
current: int,
longest: int,
last_date: date,
) -> None:
new_longest = max(current, longest)
cache_key = f"streak:{student_id}"
# Aggiorna Redis
await self.redis.hset(cache_key, mapping={
"current": current,
"longest": new_longest,
"last_date": last_date.isoformat(),
})
await self.redis.expire(cache_key, 86400 * 2) # TTL 2 giorni
# Aggiorna DB
await self.db.execute(
"""INSERT INTO student_streaks (student_id, current_streak, longest_streak, last_activity_date)
VALUES (:sid, :cur, :lon, :dt)
ON CONFLICT (student_id) DO UPDATE
SET current_streak = :cur, longest_streak = :lon, last_activity_date = :dt""",
{"sid": student_id, "cur": current, "lon": new_longest, "dt": last_date},
)
await self.db.commit()
async def _update_freeze(self, student_id: str, new_freeze: int) -> None:
await self.redis.hset(f"streak:{student_id}", "freeze", new_freeze)
await self.db.execute(
"UPDATE student_streaks SET freeze_remaining = :fr WHERE student_id = :sid",
{"sid": student_id, "fr": new_freeze},
)
await self.db.commit()Anti-modele de evitat
- Clasament global unic: Promovați 80% dintre utilizatorii care nu sunt în top 10. Folosiți segmente peer-to-peer.
- Insigne fără sens: Insignele pentru fiecare acțiune banală umflă sistemul. Insignele trebuie să reprezinte realizări reale.
- XP fără limită de viteză: Boții pot acumula XP infinit. Implementați limitarea ratei și detectarea anomaliilor.
- Stare implicită: Gestionarea progresiei cu steaguri booleene împrăștiate în întregul cod provoacă erori care sunt dificil de depanat. Utilizați întotdeauna o mașină de stări explicite.
- Atribuirea insignei fără idempotenta: Fără constrângerile UNICE și INSERT ON CONFLICT, reîncercările pot acorda aceeași insignă de mai multe ori.
- Serie fără fus orar: Un utilizator din Asia nu ar trebui să piardă o serie din cauza problemei de compensare UTC.
Concluzii și pașii următori
Am construit fundația unui motor robust de gamification: mașini cu stări explicite pentru progresie, sistem XP cu curbe logaritmice, insigne idempotente cu reguli clasamente configurabile, în timp real, peer-to-peer, cu Redis și urmărirea seriei cu suport de fus orar. Fiecare componentă este proiectată pentru a se extinde la milioane de utilizatori fără degradarea performanței.
Cheia succesului a fost abordarea configurabil: regulile insignelor sunt definite ca date JSON, nu cod codificat. Acest lucru permite managerilor de conținut pentru a crea noi mecanici de gamification fără a implica dezvoltatorii.
În următorul articol vom explora Conducta de Learning Analytics cu xAPI și Kafka: cum să colectați într-un fel date comportamentale de la elevi standardizate și transformați-le în perspective acționabile pentru profesori și administratori.
Seria EdTech Engineering
- Arhitectură LMS scalabilă: model multi-chiriași
- Algoritmi de învățare adaptivă: de la teorie la producție
- Streaming video pentru educație: WebRTC vs HLS vs DASH
- Sisteme de supraveghere AI: confidențialitate-în primul rând cu computer Vision
- Tutor personalizat cu LLM: RAG pentru fundamentarea cunoștințelor
- Motor de gamification: arhitectură și mașină de stat (acest articol)
- Learning Analytics: Data Pipeline cu xAPI și Kafka
- Colaborare în timp real în EdTech: CRDT și WebSocket
- Mobile-First EdTech: Offline-First Architecture
- Managementul conținutului cu mai mulți chiriași: Versiune și SCORM