""" XP Engine v2 — Core gamification logic with trust tiers and anti-cheat integration. Trust Tier Rules: Tier 1 (API-verified): 100% XP, auto-awarded immediately Tier 2 (Admin-imported): 100% XP, auto-awarded on import Tier 3 (Self-reported): 0% XP until admin approves → then 100% Split (Content): xp_open_pct% on file access, remainder on quiz pass Anti-cheat runs BEFORE XP is awarded. If flagged, XP is held. """ from uuid import UUID from sqlalchemy import select, update from sqlalchemy.ext.asyncio import AsyncSession from app.models.activity import Activity, ActivityCompletion from app.models.anti_cheat import VerificationRequest from app.models.gamification import ActivityFeed, Level from app.models.user import User async def get_level_for_xp(db: AsyncSession, xp: int) -> Level | None: result = await db.execute( select(Level).where(Level.min_xp <= xp).order_by(Level.min_xp.desc()).limit(1) ) return result.scalar_one_or_none() async def get_next_level(db: AsyncSession, current_xp: int) -> Level | None: result = await db.execute( select(Level).where(Level.min_xp > current_xp).order_by(Level.min_xp.asc()).limit(1) ) return result.scalar_one_or_none() async def process_completion( db: AsyncSession, user_id: UUID, activity_id: UUID, completion_id: UUID, source: str = "manual", trust_tier: int = 3, ) -> list[dict]: """ Main entry point: process an activity completion through the full pipeline. Returns events (xp_awarded, level_up, badge_earned) for frontend toasts. """ events: list[dict] = [] activity = await db.get(Activity, activity_id) if not activity: return events completion = await db.get(ActivityCompletion, completion_id) if not completion: return events user = await db.get(User, user_id) if not user: return events # Step 1: Determine effective trust tier effective_tier = min(trust_tier, activity.trust_tier) completion.trust_tier = effective_tier # Step 2: If Tier 3 (self-reported), hold XP and create verification request if effective_tier == 3: completion.status = "pending_verification" completion.xp_awarded = 0 # Create verification request db.add(VerificationRequest( completion_id=completion.id, user_id=user_id, evidence_url=completion.evidence_url, evidence_type="self_reported", notes=f"Self-reported completion of '{activity.title}'", )) db.add(ActivityFeed( user_id=user_id, event_type="completion", event_data={ "activity_title": activity.title, "status": "pending_verification", "trust_tier": 3, }, )) return events # No XP yet # Step 3: Run anti-cheat checks (Tier 1 and 2) from app.services.anti_cheat import run_anti_cheat_checks flags = await run_anti_cheat_checks(db, user_id, activity_id, completion) critical_flags = [f for f in flags if f.severity == "critical"] if critical_flags: completion.status = "flagged" completion.flagged_reason = critical_flags[0].description completion.xp_awarded = 0 return events # XP held until admin resolves # Step 4: Calculate and award XP # For split content (reading/video with quiz), only award xp_open_pct here # The rest comes from quiz completion via content_tracker service if activity.xp_open_pct < 100 and completion.source != "quiz": xp_amount = int(activity.xp_value * activity.xp_open_pct / 100) else: xp_amount = activity.xp_value old_xp = user.total_xp new_xp = old_xp + xp_amount # Update user XP update_values = {"total_xp": new_xp} if effective_tier <= 2: update_values["verified_xp"] = User.verified_xp + xp_amount await db.execute(update(User).where(User.id == user_id).values(**update_values)) # Update completion completion.xp_awarded = xp_amount completion.source_detail = f"{source}_tier{effective_tier}" events.append({ "type": "xp_awarded", "user_id": str(user_id), "data": { "xp_amount": xp_amount, "new_total": new_xp, "activity_title": activity.title, "trust_tier": effective_tier, }, }) # Step 5: Check level-up old_level = await get_level_for_xp(db, old_xp) new_level = await get_level_for_xp(db, new_xp) if new_level and (not old_level or new_level.name != old_level.name): await db.execute( update(User).where(User.id == user_id).values(current_level=new_level.name) ) events.append({ "type": "level_up", "user_id": str(user_id), "data": { "old_level": old_level.name if old_level else "None", "new_level": new_level.name, "new_level_color": new_level.color, }, }) db.add(ActivityFeed( user_id=user_id, event_type="level_up", event_data={"old_level": old_level.name if old_level else "None", "new_level": new_level.name}, )) # Step 6: Log to activity feed db.add(ActivityFeed( user_id=user_id, event_type="completion", event_data={ "activity_id": str(activity_id), "activity_title": activity.title, "activity_category": activity.category, "xp_awarded": xp_amount, "trust_tier": effective_tier, }, )) # Step 7: Evaluate badges from app.services.badge_engine import evaluate_badges badge_events = await evaluate_badges(db, user_id) events.extend(badge_events) return events async def approve_verification( db: AsyncSession, completion_id: UUID, reviewer_id: UUID, ) -> list[dict]: """ Admin approves a pending verification. Award the held XP. """ completion = await db.get(ActivityCompletion, completion_id) if not completion or completion.status != "pending_verification": return [] completion.status = "verified" completion.verified_by = reviewer_id completion.trust_tier = 2 # Upgraded from Tier 3 to Tier 2 on approval # Now process as a normal completion with Tier 2 return await process_completion( db, completion.user_id, completion.activity_id, completion.id, source="verified", trust_tier=2, ) async def reject_verification( db: AsyncSession, completion_id: UUID, reviewer_id: UUID, reason: str = "", ) -> None: """Admin rejects a pending verification.""" completion = await db.get(ActivityCompletion, completion_id) if not completion: return completion.status = "not_started" completion.verified_by = reviewer_id completion.verification_note = reason completion.xp_awarded = 0 async def resolve_anomaly_flag( db: AsyncSession, flag_id: UUID, reviewer_id: UUID, approve: bool, note: str = "", ) -> list[dict]: """ Admin resolves an anomaly flag. If approved: release held XP. If rejected: revoke. """ from datetime import datetime, timezone from app.models.anti_cheat import AnomalyFlag flag = await db.get(AnomalyFlag, flag_id) if not flag or flag.resolved: return [] flag.resolved = True flag.resolved_by = reviewer_id flag.resolved_at = datetime.now(timezone.utc) flag.resolution_note = note if not flag.completion_id: return [] completion = await db.get(ActivityCompletion, flag.completion_id) if not completion or completion.status != "flagged": return [] if approve: completion.status = "completed" completion.flagged_reason = None return await process_completion( db, completion.user_id, completion.activity_id, completion.id, source="anomaly_resolved", trust_tier=completion.trust_tier, ) else: completion.status = "not_started" completion.xp_awarded = 0 completion.verification_note = f"Rejected: {note}" return []