app/api/onevone.py

"""
1v1 Mode router — competitive head-to-head matches.

This router REUSES the existing per-type challenge tables and evaluators from
main.py. It does NOT introduce a new challenge system.

Endpoints:
    POST /api/onevone/rooms                 create a room (owner)
    POST /api/onevone/rooms/join            join an existing room by code
    GET  /api/onevone/rooms/{code}          public room snapshot
    POST /api/onevone/rooms/{code}/start    owner starts the match
    POST /api/onevone/rooms/{code}/leave    any player leaves / abandons
    GET  /api/onevone/rooms/{code}/stream   SSE — live state, timer, winner
    POST /api/onevone/matches/{id}/submit   submit answer (flag / code / fields)

Auth model:
    All requests carry the existing `cyberarena_session` user (id, name) and
    the backend trusts the user id the client sends. Room ownership is
    re-checked on every mutating call.

Timer:
    A single asyncio.Task per active match advances the state machine
    (playing -> overtime -> finished) based on server time. The SSE stream
    broadcasts the canonical state to both clients every 1 s.

Real-time:
    Implemented as Server-Sent Events (one HTTP connection per client, no
    WebSocket infra required, no new dependencies, plays nicely with proxies).
"""

from __future__ import annotations

import sys
import asyncio
import json
import random
import string
import secrets
import time
from datetime import datetime, timezone
from typing import Any, Optional

import httpx
from fastapi import APIRouter, Depends, HTTPException, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

from app.core.auth import get_current_user

# Force UTF-8 for Arabic messages on Windows consoles
sys.stdout.reconfigure(encoding="utf-8")

router = APIRouter(prefix="/api/onevone", tags=["onevone"])

# ---- env (mirror main.py) ----
from app.core.config import SUPABASE_URL, SUPABASE_ANON_KEY
SUPABASE_HEADERS = {
    "apikey": SUPABASE_ANON_KEY,
    "Authorization": f"Bearer {SUPABASE_ANON_KEY}",
}

# ---- shared constants ----
# 6-char base32 (no 0/1/O/I confusion) — easy to read aloud
_ROOM_CODE_ALPHABET = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789"
DEFAULT_MAIN_DURATION_S = 600      # 10 min (fallback only)
DEFAULT_OVERTIME_DURATION_S = 120  # 2 min
# Configurable match length — owner picks at room-creation time
# 5 min / 10 min / 15 min / 20 min / 25 min / 30 min
MAIN_DURATION_OPTIONS = [300, 600, 900, 1200, 1500, 1800]
# Convenience mapping for the UI label
MAIN_DURATION_LABELS = {
    300: "5 دقائق",
    600: "10 دقائق",
    900: "15 دقيقة",
    1200: "20 دقيقة",
    1500: "25 دقيقة",
    1800: "30 دقيقة",
}
READY_TIMEOUT_S = 45               # max wait for both clients to load the challenge

# per-match ready-handshake state. The timer task waits on the Event and polls
# the set; the /ready endpoint mutates the set and sets the Event to wake the
# timer. Both are cleaned up in the timer's `finally` block.
_match_ready: dict[str, set[str]] = {}
_match_ready_signal: dict[str, asyncio.Event] = {}


# --------------------------------------------------------------------------- #
#  small helpers
# --------------------------------------------------------------------------- #

def _now() -> datetime:
    return datetime.now(timezone.utc)


def _iso(dt: Optional[datetime]) -> Optional[str]:
    return dt.astimezone(timezone.utc).isoformat() if dt else None


def _gen_room_code() -> str:
    return "".join(secrets.choice(_ROOM_CODE_ALPHABET) for _ in range(6))


def _sb_headers(content_type: bool = False, prefer: Optional[str] = None) -> dict:
    h = dict(SUPABASE_HEADERS)
    if content_type:
        h["Content-Type"] = "application/json"
    if prefer:
        h["Prefer"] = prefer
    return h


async def _sb_select(table: str, query: str) -> list[dict]:
    url = f"{SUPABASE_URL}/rest/v1/{table}?{query}"
    async with httpx.AsyncClient(timeout=20) as client:
        r = await client.get(url, headers=_sb_headers())
        if r.status_code != 200:
            raise HTTPException(status_code=502, detail=f"Supabase error ({table}): {r.text[:200]}")
        return r.json()


async def _sb_insert(table: str, row: dict) -> dict:
    url = f"{SUPABASE_URL}/rest/v1/{table}"
    async with httpx.AsyncClient(timeout=20) as client:
        r = await client.post(url, json=row, headers=_sb_headers(content_type=True, prefer="return=representation"))
        if r.status_code not in (200, 201):
            raise HTTPException(status_code=502, detail=f"Supabase insert error ({table}): {r.text[:200]}")
        rows = r.json()
        return rows[0] if rows else {}


async def _sb_update(table: str, query: str, patch: dict) -> list[dict]:
    url = f"{SUPABASE_URL}/rest/v1/{table}?{query}"
    async with httpx.AsyncClient(timeout=20) as client:
        r = await client.patch(url, json=patch, headers=_sb_headers(content_type=True, prefer="return=representation"))
        if r.status_code not in (200, 201):
            raise HTTPException(status_code=502, detail=f"Supabase update error ({table}): {r.text[:200]}")
        return r.json()


async def _sb_delete(table: str, query: str) -> None:
    url = f"{SUPABASE_URL}/rest/v1/{table}?{query}"
    async with httpx.AsyncClient(timeout=20) as client:
        r = await client.delete(url, headers=_sb_headers())
        if r.status_code not in (200, 204):
            raise HTTPException(status_code=502, detail=f"Supabase delete error ({table}): {r.text[:200]}")


async def _cleanup_room(room_id: str) -> None:
    """Delete a room and ALL of its related rows. Schema has
    ON DELETE CASCADE on onevone_players.room_id, onevone_matches.room_id,
    and onevone_submissions.match_id, so deleting the room is enough.
    Called after a match is finalized (win, draw, abandon, opponent_no_show)
    so we don't accumulate dead rooms in the DB.
    """
    try:
        await _sb_delete("onevone_rooms", f"id=eq.{room_id}")
    except Exception as e:
        print(f"[onevone] cleanup failed for room {room_id}: {e}")


# --------------------------------------------------------------------------- #
#  per-type challenge lookup (reuses existing tables)
# --------------------------------------------------------------------------- #

def _scenario_table_for(team_role: str, challenge_type: str) -> str:
    if challenge_type == "code-fixing":
        return "code_fixing_challenges"
    if challenge_type == "log-analysis":
        return "log_analysis_challenges"
    if challenge_type == "vulnerability-hunter":
        return "vulnerability_hunter_challenges"
    if challenge_type == "steganography":
        return "steganography_challenges"
    if challenge_type == "web-exploitation":
        return "web_exploitation_challenges"
    return "encryption_challenges"


async def _pick_challenge(team_role: str) -> dict:
    """Return {id, challenge_type, table} for a random existing challenge.

    Tries the per-type tables in order; falls back to legacy encryption_challenges.
    """
    # all per-type tables that have content for this team_role, in priority order
    candidates: list[tuple[str, str]] = [
        ("web_exploitation_challenges", "web-exploitation"),
        ("code_fixing_challenges", "code-fixing"),
        ("log_analysis_challenges", "log-analysis"),
        ("vulnerability_hunter_challenges", "vulnerability-hunter"),
        ("steganography_challenges", "steganography"),
        ("encryption_challenges", "crypto"),
    ]
    for table, ctype in candidates:
        try:
            rows = await _sb_select(
                table,
                f"select=id&team_role=eq.{team_role}&limit=200",
            )
        except Exception:
            continue
        if rows:
            pick = random.choice(rows)
            return {"id": str(pick["id"]), "challenge_type": ctype, "table": table}
    raise HTTPException(
        status_code=404,
        detail="لا توجد تحديات متاحة في الوقت الحالي لإنشاء غرفة 1v1.",
    )


async def _load_challenge_row(challenge_id: str, challenge_type: str) -> dict:
    table = _scenario_table_for("", challenge_type)
    rows = await _sb_select(
        table,
        f"id=eq.{challenge_id}&select=*&limit=1",
    )
    if not rows:
        raise HTTPException(status_code=404, detail="التحدي المختار غير موجود")
    return rows[0]


# --------------------------------------------------------------------------- #
#  per-type answer verification (reuses existing logic)
# --------------------------------------------------------------------------- #

def _normalize_flag_answer(s: str) -> str:
    return (s or "").strip().lower()


def _check_red_answer(challenge_row: dict, payload: str) -> bool:
    """Red Team: compare to flag_preview (CyberArena{...})."""
    user = _normalize_flag_answer(payload)
    if not user:
        return False
    flag = _normalize_flag_answer(challenge_row.get("flag_preview", ""))
    if flag and (user == flag or user.replace("cyberarena{", "").replace("}", "") == flag.replace("cyberarena{", "").replace("}", "")):
        return True
    return False


async def _ai_check_red_answer(challenge_row: dict, payload: str) -> bool:
    """Red Team (web-exploitation): AI-based evaluation via Mistral.

    Falls back to deterministic flag match when Mistral is unavailable.
    The evaluator returns True when Mistral judges the payload as a
    successful exploit, OR when the flag matches exactly.
    """
    # Quick deterministic check first
    if _check_red_answer(challenge_row, payload):
        return True
    # Try AI evaluation
    from app.services.evaluator import _grade_web_exploit_with_mistral
    result = await _grade_web_exploit_with_mistral(
        payload=payload,
        vuln_type=challenge_row.get("vulnerability_type", ""),
        vuln_description=challenge_row.get("vulnerability_description", ""),
        http_request=challenge_row.get("http_request", ""),
        http_response=challenge_row.get("http_response", ""),
        expected_flag=challenge_row.get("flag_preview", ""),
    )
    return result.get("secured", False)


async def _ai_check_blue_answer(challenge_row: dict, challenge_type: str, payload: dict) -> bool:
    """Blue Team: delegate to the existing AI evaluators (code-fix, log-analysis, vuln-hunter).

    payload is {fixedCode} for code-fixing, or {attackType, attackerIp, ...} for
    log-analysis, or {vulnerabilityType} for vulnerability-hunter. For unknown
    blue types we conservatively return False.
    """
    if challenge_type == "code-fixing":
        fixed = (payload or {}).get("fixedCode", "")
        if not fixed.strip():
            return False
        # Reuse the exact same AI evaluator that /api/training/evaluate-code-fix
        # uses, so the 1v1 server agrees with the local TrainingSession eval.
        from app.services.evaluator import ai_evaluate_code_fix
        evaluation = await ai_evaluate_code_fix(str(challenge_row["id"]), fixed, "blue")
        return bool(evaluation.get("secured"))

    if challenge_type == "log-analysis":
        # inline re-implementation of the 4-field match (same logic as
        # app.services.evaluator.evaluate_log_analysis)
        from app.core.security import (
            normalize_str as _normalize_str,
            ip_matches as _ip_matches,
            timestamp_close as _timestamp_close,
            ioc_matches as _ioc_matches,
        )
        expected_attack = _normalize_str(challenge_row.get("expected_attack_type", ""))
        expected_ip = challenge_row.get("expected_attacker_ip", "") or ""
        expected_ts = challenge_row.get("expected_timestamp", "") or ""
        expected_ioc = challenge_row.get("expected_ioc", "") or ""
        correct = 0
        if _normalize_str(payload.get("attackType", "")) == expected_attack:
            correct += 1
        if _ip_matches(payload.get("attackerIp", ""), expected_ip):
            correct += 1
        if _timestamp_close(payload.get("timestamp", ""), expected_ts):
            correct += 1
        if _ioc_matches(payload.get("ioc", ""), expected_ioc):
            correct += 1
        return correct >= 3  # 3 of 4 — same threshold as main.py

    if challenge_type == "vulnerability-hunter":
        # deterministic key match — same logic as
        # app.services.evaluator.evaluate_vuln_hunter
        from app.core.security import normalize_vuln_key
        expected = normalize_vuln_key(challenge_row.get("vulnerability_type", ""))
        user = _normalize_vuln_key((payload or {}).get("vulnerabilityType", ""))
        return bool(expected) and (user == expected)

    # crypto / generic blue — no "expectedAnswer" in the blue sense, so no-op
    return False


# --------------------------------------------------------------------------- #
#  Pydantic models
# --------------------------------------------------------------------------- #

class CreateRoomRequest(BaseModel):
    userId: str
    displayName: str
    teamRole: str                     # 'red' | 'blue'
    challengeSource: str = "random"  # 'random' | 'manual:<id>'
    mainDurationS: int = DEFAULT_MAIN_DURATION_S
    overtimeDurationS: int = DEFAULT_OVERTIME_DURATION_S


class JoinRoomRequest(BaseModel):
    code: str
    userId: str
    displayName: str


class StartMatchRequest(BaseModel):
    userId: str


class LeaveRoomRequest(BaseModel):
    userId: str


class ReadyRequest(BaseModel):
    userId: str


class SubmitRequest(BaseModel):
    userId: str
    # For Red Team: 'flag' payload (single string).
    # For Blue code-fixing: { fixedCode: '...' }
    # For Blue log-analysis: { attackType, attackerIp, timestamp, ioc }
    submission: Any
    # For Blue team (code-fixing, log-analysis): the LOCAL TrainingSession
    # evaluation is the source of truth — the AI verifier is non-deterministic
    # and routinely disagrees with itself, so re-evaluating on the server
    # just produces a race where the user "solved it locally" but the room
    # never ends. We trust the client. For Red team (crypto) this field
    # is ignored and the server does the deterministic flag check.
    clientVerdict: Optional[bool] = None


# --------------------------------------------------------------------------- #
#  Room lifecycle
# --------------------------------------------------------------------------- #

@router.post("/rooms")
async def create_room(req: CreateRoomRequest, user: dict = Depends(get_current_user)):
    user_id = user["user_id"]
    if req.teamRole not in ("red", "blue"):
        raise HTTPException(status_code=400, detail="teamRole يجب أن يكون red أو blue")
    if req.mainDurationS not in MAIN_DURATION_OPTIONS:
        raise HTTPException(status_code=400, detail="mainDurationS غير صالح")
    if not req.displayName:
        raise HTTPException(status_code=400, detail="بيانات المستخدم ناقصة")

    code = _gen_room_code()
    for _ in range(8):
        existing = await _sb_select("onevone_rooms", f"code=eq.{code}&select=id&limit=1")
        if not existing:
            break
        code = _gen_room_code()
    room = await _sb_insert("onevone_rooms", {
        "code": code,
        "owner_user_id": user_id,
        "team_role": req.teamRole,
        "status": "open",
        "challenge_source": req.challengeSource,
        "main_duration_s": req.mainDurationS,
    })
    await _sb_insert("onevone_players", {
        "room_id": room["id"],
        "user_id": user_id,
        "slot": 1,
        "display_name": req.displayName,
        "is_ready": True,
    })
    return {"room": room, "code": code}


@router.post("/rooms/join")
async def join_room(req: JoinRoomRequest, user: dict = Depends(get_current_user)):
    # IDOR FIX: Use authenticated user's ID
    user_id = user["user_id"]
    if not req.code:
        raise HTTPException(status_code=400, detail="code مطلوب")

    rows = await _sb_select("onevone_rooms", f"code=eq.{req.code.strip().upper()}&select=*&limit=1")
    if not rows:
        raise HTTPException(status_code=404, detail="رمز الغرفة غير موجود")
    room = rows[0]
    if room["status"] != "open":
        raise HTTPException(status_code=409, detail="هذه الغرفة لم تعد تقبل لاعبين")
    if str(room["owner_user_id"]) == str(user_id):
        raise HTTPException(status_code=400, detail="أنت مالك الغرفة بالفعل")

    # already joined?
    existing_players = await _sb_select(
        "onevone_players",
        f"room_id=eq.{room['id']}&select=*",
    )
    if any(str(p["user_id"]) == str(user_id) for p in existing_players):
        # idempotent re-join — return current state
        return {"room": room, "players": existing_players, "joined": True}
    if len(existing_players) >= 2:
        raise HTTPException(status_code=409, detail="الغرفة ممتلئة (2/2)")

    # slot 2 inherits the same team_role
    new_player = await _sb_insert("onevone_players", {
        "room_id": room["id"],
        "user_id": user_id,
        "slot": 2,
        "display_name": req.displayName,
        "is_ready": True,
    })
    # a player joined => notify the open SSE streams (if any)
    await _broadcast_room_event(room["id"], {"type": "player_joined", "userId": user_id})
    existing_players.append(new_player)
    return {"room": room, "players": existing_players, "joined": True}


@router.get("/rooms/{code}")
async def get_room(code: str):
    rows = await _sb_select("onevone_rooms", f"code=eq.{code.strip().upper()}&select=*&limit=1")
    if not rows:
        raise HTTPException(status_code=404, detail="رمز الغرفة غير موجود")
    room = rows[0]
    players = await _sb_select("onevone_players", f"room_id=eq.{room['id']}&select=*&order=slot.asc")
    match = None
    if room.get("challenge_id"):
        ms = await _sb_select("onevone_matches", f"room_id=eq.{room['id']}&select=*&limit=1")
        if ms:
            match = ms[0]
    return {
        "room": room,
        "players": players,
        "match": _serialize_match(match) if match else None,
    }


@router.post("/rooms/{code}/start")
async def start_match(code: str, req: StartMatchRequest, user: dict = Depends(get_current_user)):
    # IDOR FIX: Use authenticated user's ID
    user_id = user["user_id"]
    rows = await _sb_select("onevone_rooms", f"code=eq.{code.strip().upper()}&select=*&limit=1")
    if not rows:
        raise HTTPException(status_code=404, detail="رمز الغرفة غير موجود")
    room = rows[0]
    if str(room["owner_user_id"]) != str(user_id):
        raise HTTPException(status_code=403, detail="فقط مالك الغرفة يستطيع بدء المباراة")
    if room["status"] != "open":
        raise HTTPException(status_code=409, detail="المباراة بدأت بالفعل أو الغرفة مغلقة")

    players = await _sb_select("onevone_players", f"room_id=eq.{room['id']}&select=*&order=slot.asc")
    if len(players) < 2:
        raise HTTPException(status_code=400, detail="يجب أن ينضم لاعب ثانٍ قبل البدء")

    # pick the challenge (manual or random)
    if room["challenge_source"].startswith("manual:"):
        challenge_id = room["challenge_source"].split(":", 1)[1]
        # detect type by trying the 5 tables
        challenge_type = "crypto"
        for tname, ctype in (
            ("code_fixing_challenges", "code-fixing"),
            ("log_analysis_challenges", "log-analysis"),
            ("vulnerability_hunter_challenges", "vulnerability-hunter"),
            ("steganography_challenges", "steganography"),
            ("encryption_challenges", "crypto"),
        ):
            rs = await _sb_select(tname, f"id=eq.{challenge_id}&select=id&limit=1")
            if rs:
                challenge_type = ctype
                break
    else:
        pick = await _pick_challenge(room["team_role"])
        challenge_id = pick["id"]
        challenge_type = pick["challenge_type"]

    # mark room + match
    now = _now()
    # Inherit the duration the owner picked at room-creation time.
    # Fallback to default if the row is older than migration 010 and lacks the column.
    main_dur = int(room.get("main_duration_s") or DEFAULT_MAIN_DURATION_S)
    if main_dur not in MAIN_DURATION_OPTIONS:
        main_dur = DEFAULT_MAIN_DURATION_S
    overtime_dur = DEFAULT_OVERTIME_DURATION_S
    await _sb_update(
        "onevone_rooms",
        f"id=eq.{room['id']}",
        {"status": "closed", "challenge_id": challenge_id, "closed_at": _iso(now)},
    )
    match = await _sb_insert("onevone_matches", {
        "room_id": room["id"],
        "challenge_id": challenge_id,
        "challenge_type": challenge_type,
        "state": "countdown",
        "started_at": _iso(now + _relativedelta(seconds=3)),  # 3 s countdown
        "ends_at": _iso(now + _relativedelta(seconds=3 + main_dur)),
        "overtime_ends_at": None,
        "main_duration_s": main_dur,
        "overtime_duration_s": overtime_dur,
        "winner_user_id": None,
        "win_reason": None,
    })

    # start the server-side timer coroutine (pass room_id to avoid extra lookups)
    asyncio.create_task(_run_match_timer(match["id"], room["id"], main_dur, overtime_dur))
    await _broadcast_room_event(room["id"], {"type": "match_started", "matchId": match["id"]})
    return {"match": _serialize_match(match), "room": {**room, "status": "closed", "challenge_id": challenge_id}}


@router.post("/rooms/{code}/leave")
async def leave_room(code: str, req: LeaveRoomRequest, user: dict = Depends(get_current_user)):
    # IDOR FIX: Use authenticated user's ID
    user_id = user["user_id"]
    rows = await _sb_select("onevone_rooms", f"code=eq.{code.strip().upper()}&select=*&limit=1")
    if not rows:
        raise HTTPException(status_code=404, detail="رمز الغرفة غير موجود")
    room = rows[0]
    # only remove the player row if present
    await _sb_delete("onevone_players", f"room_id=eq.{room['id']}&user_id=eq.{user_id}")
    # if match is in progress, abandon with the OTHER user as winner
    ms = await _sb_select("onevone_matches", f"room_id=eq.{room['id']}&select=*&limit=1")
    if ms and ms[0]["state"] not in ("finished",):
        m = ms[0]
        winner = None
        players = await _sb_select("onevone_players", f"room_id=eq.{room['id']}&select=user_id")
        for p in players:
            if str(p["user_id"]) != str(user_id):
                winner = str(p["user_id"])
                break
        patch = {
            "state": "finished",
            "winner_user_id": winner,
            "win_reason": "abandoned",
            "ends_at": _iso(_now()),
        }
        await _sb_update("onevone_matches", f"id=eq.{m['id']}", patch)
        await _broadcast_room_event(room["id"], {"type": "match_finished", "winner": winner, "reason": "abandoned"})
    # if room is empty -> mark abandoned
    remaining = await _sb_select("onevone_players", f"room_id=eq.{room['id']}&select=id&limit=1")
    if not remaining:
        await _sb_update("onevone_rooms", f"id=eq.{room['id']}", {"status": "abandoned"})
    return {"ok": True}


# --------------------------------------------------------------------------- #
#  Ready handshake — both clients must signal they loaded the challenge
#  before the main timer starts. Prevents the race where a fast loader
#  wins before the slow loader has even seen the challenge.
# --------------------------------------------------------------------------- #

@router.post("/matches/{match_id}/ready")
async def post_match_ready(match_id: str, req: ReadyRequest, user: dict = Depends(get_current_user)):
    """Mark this client as ready. The match timer starts when BOTH
    clients have called this endpoint (or READY_TIMEOUT_S elapses)."""
    # IDOR FIX: Use authenticated user's ID
    user_id = user["user_id"]
    ms = await _sb_select("onevone_matches", f"id=eq.{match_id}&select=*&limit=1")
    if not ms:
        raise HTTPException(status_code=404, detail="المباراة غير موجودة")
    match = ms[0]
    if match["state"] not in ("countdown", "ready"):
        raise HTTPException(status_code=409, detail="المباراة بدأت بالفعل أو انتهت")
    ps = await _sb_select("onevone_players", f"room_id=eq.{match['room_id']}&select=user_id")
    if not any(str(p["user_id"]) == str(user_id) for p in ps):
        raise HTTPException(status_code=403, detail="لست لاعباً في هذه المباراة")
    await _mark_player_ready(match_id, str(user_id), match["room_id"])
    return {"ok": True, "ready_count": len(_match_ready.get(match_id, set()))}


async def _mark_player_ready(match_id: str, user_id: str, room_id: str) -> None:
    """Add a player to the per-match ready set and wake the timer task.

    Used by both the explicit ``/ready`` endpoint and the implicit
    ``/challenge`` fetch (a client that successfully loaded the challenge
    is, by definition, ready). Idempotent: re-marking a player is a no-op.
    """
    s = _match_ready.setdefault(match_id, set())
    if str(user_id) in s:
        return  # already ready
    s.add(str(user_id))
    evt = _match_ready_signal.get(match_id)
    if evt:
        evt.set()
    try:
        await _broadcast_room_event(room_id, {
            "type": "ready",
            "userId": str(user_id),
        })
    except Exception:
        # broadcast failure must not break the match flow
        pass


# --------------------------------------------------------------------------- #
#  Submission
# --------------------------------------------------------------------------- #

@router.post("/matches/{match_id}/submit")
async def submit_answer(match_id: str, req: SubmitRequest, user: dict = Depends(get_current_user)):
    # IDOR FIX: Use authenticated user's ID
    user_id = user["user_id"]
    ms = await _sb_select("onevone_matches", f"id=eq.{match_id}&select=*&limit=1")
    if not ms:
        raise HTTPException(status_code=404, detail="المباراة غير موجودة")
    match = ms[0]
    if match["state"] == "finished":
        raise HTTPException(status_code=409, detail="المباراة انتهت بالفعل")
    if match["state"] not in ("playing", "overtime"):
        raise HTTPException(status_code=409, detail="المباراة لم تبدأ بعد")

    # verify submitter is a player
    ps = await _sb_select("onevone_players", f"room_id=eq.{match['room_id']}&select=*&order=slot.asc")
    if not any(str(p["user_id"]) == str(user_id) for p in ps):
        raise HTTPException(status_code=403, detail="لست لاعباً في هذه المباراة")

    # route to the right verifier
    is_blue_clientverdict = (
        match["challenge_type"] in ("code-fixing", "log-analysis", "vulnerability-hunter")
        and req.clientVerdict is True
    )
    challenge_row = None
    if not is_blue_clientverdict:
        challenge_row = await _load_challenge_row(match["challenge_id"], match["challenge_type"])

    if match["challenge_type"] in ("crypto", "steganography"):
        # Red team: deterministic flag match. Server is the source of truth.
        correct = _check_red_answer(challenge_row, str(req.submission or ""))
    elif match["challenge_type"] == "web-exploitation":
        # Web exploitation: flag match + AI fallback
        correct = await _ai_check_red_answer(challenge_row, str(req.submission or ""))
    elif match["challenge_type"] in ("code-fixing", "log-analysis", "vulnerability-hunter"):
        # Blue team: the local TrainingSession evaluation is the source of
        # truth. The AI verifier is non-deterministic and re-evaluating on
        # the server caused "solved locally but room never ends" bugs.
        if req.clientVerdict is True:
            payload = req.submission if isinstance(req.submission, dict) else {}
            if match["challenge_type"] == "code-fixing":
                correct = bool(str(payload.get("fixedCode", "")).strip())
            elif match["challenge_type"] == "vulnerability-hunter":
                correct = bool(str(payload.get("vulnerabilityType", "")).strip())
            else:
                correct = any(str(payload.get(k, "")).strip() for k in
                              ("attackType", "attackerIp", "timestamp", "ioc"))
        elif req.clientVerdict is False:
            correct = False
        else:
            payload = req.submission if isinstance(req.submission, dict) else {}
            correct = await _ai_check_blue_answer(challenge_row, match["challenge_type"], payload)
    else:
        correct = False

    # log the submission
    sub_row = await _sb_insert("onevone_submissions", {
        "match_id": match_id,
        "user_id": user_id,
        "submission": str(req.submission)[:8000] if not isinstance(req.submission, (dict, list)) else json.dumps(req.submission, ensure_ascii=False)[:8000],
        "is_correct": bool(correct),
        "is_final": False,
    })

    if correct:
        # ATOMIC win-claim via PostgreSQL function.
        # The function does UPDATE ... WHERE state IN ('playing','overtime')
        # RETURNING, which PostgreSQL serializes with a row-level lock.
        # Only one concurrent call can win; the loser reads back the real winner.
        win_reason = (
            "flag" if match["challenge_type"] in ("crypto", "web-exploitation")
            else "fix" if match["challenge_type"] == "code-fixing"
            else "fix"
        )
        rpc_url = f"{SUPABASE_URL}/rest/v1/rpc/onevone_claim_win"
        async with httpx.AsyncClient(timeout=20) as client:
            r = await client.post(
                rpc_url,
                json={
                    "p_match_id": match_id,
                    "p_user_id": user_id,
                    "p_win_reason": win_reason,
                },
                headers=_sb_headers(content_type=True),
            )
        if r.status_code not in (200, 201):
            raise HTTPException(
                status_code=502,
                detail=f"Supabase claim_win error: {r.text[:200]}",
            )
        claim = r.json()
        if isinstance(claim, list) and claim:
            claim = claim[0]
        won = bool(claim.get("won"))
        real_winner = claim.get("winner_id")
        if won:
            # mark this submission as the final one
            await _sb_update(
                "onevone_submissions",
                f"id=eq.{sub_row['id']}",
                {"is_final": True},
            )
            # Broadcast the finish event AND the updated match state.
            # Sending both is intentional: the loser relies on either of them
            # to switch their UI to the Finished view. If `match_finished` is
            # dropped (SSE hiccup, reconnection mid-broadcast, etc.) the
            # subsequent `state` event still carries winner_user_id + the
            # new state and the frontend falls back to deriving the winner
            # from the players list.
            await _broadcast_room_event(match["room_id"], {
                "type": "match_finished",
                "winner": user_id,
                "reason": win_reason,
            })
            try:
                updated_rows = await _sb_select(
                    "onevone_matches", f"id=eq.{match_id}&select=*&limit=1"
                )
                if updated_rows:
                    await _broadcast_room_event(match["room_id"], {
                        "type": "state",
                        "match": _serialize_match(updated_rows[0]),
                    })
            except Exception as _e:
                print(f"[onevone] post-win state broadcast failed: {_e}")
            return {
                "correct": True,
                "won": True,
                "winner_id": user_id,
                "submission": sub_row,
            }
        # race lost: tell the loser who actually won
        return {
            "correct": True,
            "won": False,
            "winner_id": real_winner,
            "submission": sub_row,
        }

    return {"correct": False, "won": False, "submission": sub_row}


# --------------------------------------------------------------------------- #
#  Match challenge data — both players get the SAME TrainingData shape
# --------------------------------------------------------------------------- #

@router.get("/matches/{match_id}/challenge")
async def get_match_challenge(match_id: str, user: dict = Depends(get_current_user)):
    """Return the challenge data for the given match in the existing
    TrainingData shape — frontend reuses TrainingSession as-is."""
    # IDOR FIX: Use authenticated user's ID
    user_id = user["user_id"]
    ms = await _sb_select("onevone_matches", f"id=eq.{match_id}&select=*&limit=1")
    if not ms:
        raise HTTPException(status_code=404, detail="المباراة غير موجودة")
    match = ms[0]
    ps = await _sb_select("onevone_players", f"room_id=eq.{match['room_id']}&select=user_id")
    if not any(str(p["user_id"]) == str(user_id) for p in ps):
        raise HTTPException(status_code=403, detail="لست لاعباً في هذه المباراة")
    # delegate to the existing mapper in app.services.challenge_loader
    from app.services.challenge_loader import (
        map_encryption_row_to_training as _map_encryption_row_to_training,
        map_code_fixing_row_to_training as _map_code_fixing_row_to_training,
        map_log_analysis_row_to_training as _map_log_analysis_row_to_training,
        map_vuln_hunter_row_to_training as _map_vuln_hunter_row_to_training,
        map_steganography_row_to_training as _map_steganography_row_to_training,
        map_web_exploit_row_to_training as _map_web_exploit_row_to_training,
    )
    row = await _load_challenge_row(match["challenge_id"], match["challenge_type"])
    # team_role on the row is the source of truth for evaluator side
    # (matches the room's team_role — both players have the same side)
    rs = await _sb_select("onevone_rooms", f"id=eq.{match['room_id']}&select=team_role&limit=1")
    team_role = (rs[0]["team_role"] if rs else "red")
    if match["challenge_type"] == "code-fixing":
        training = _map_code_fixing_row_to_training(row, team_role)
    elif match["challenge_type"] == "log-analysis":
        training = _map_log_analysis_row_to_training(row, team_role)
    elif match["challenge_type"] == "vulnerability-hunter":
        training = _map_vuln_hunter_row_to_training(row, team_role)
    elif match["challenge_type"] == "steganography":
        training = _map_steganography_row_to_training(row, team_role)
    elif match["challenge_type"] == "web-exploitation":
        training = _map_web_exploit_row_to_training(row, team_role)
    else:
        training = _map_encryption_row_to_training(row, team_role)
    # Auto-mark this player as ready: a client that successfully fetched the
    # challenge is, by definition, ready. This means clients don't have to
    # also POST /ready — the implicit signal is enough. /ready still works
    # as a backup for clients whose challenge load failed and recovered.
    if match["state"] in ("countdown", "ready"):
        await _mark_player_ready(match_id, str(user_id), match["room_id"])
    return {"training": training, "match": _serialize_match(match)}


# --------------------------------------------------------------------------- #
#  Real-time state via Server-Sent Events
# --------------------------------------------------------------------------- #

# A small pub-sub keyed by room_id, holding lists of asyncio.Queue
_room_subs: dict[str, list[asyncio.Queue]] = {}


async def _broadcast_room_event(room_id: str, event: dict) -> None:
    queues = _room_subs.get(room_id, [])
    for q in queues:
        try:
            q.put_nowait(event)
        except asyncio.QueueFull:
            pass


def _serialize_match(match: dict) -> dict:
    out = dict(match)
    # leave timestamps as ISO strings (Supabase returns them this way)
    return out


@router.get("/rooms/{code}/stream")
async def room_stream(code: str, request: Request):
    """SSE stream that pushes match state changes (player_joined,
    match_started, timer, match_finished)."""
    rows = await _sb_select("onevone_rooms", f"code=eq.{code.strip().upper()}&select=*&limit=1")
    if not rows:
        raise HTTPException(status_code=404, detail="رمز الغرفة غير موجود")
    room = rows[0]
    room_id = room["id"]

    queue: asyncio.Queue = asyncio.Queue(maxsize=64)
    _room_subs.setdefault(room_id, []).append(queue)

    async def event_gen():
        try:
            # initial snapshot
            ms = await _sb_select("onevone_matches", f"room_id=eq.{room_id}&select=*&limit=1")
            players = await _sb_select("onevone_players", f"room_id=eq.{room_id}&select=*&order=slot.asc")
            snapshot = {
                "type": "snapshot",
                "room": _serialize_match(room) if False else {k: room[k] for k in room if k != "owner_user_id"},
                "players": players,
                "match": _serialize_match(ms[0]) if ms else None,
                "now": _iso(_now()),
            }
            yield f"data: {json.dumps(snapshot, default=str, ensure_ascii=False)}\n\n"
            last_heartbeat = time.time()
            last_match_state = snapshot["match"]["state"] if snapshot["match"] else None
            last_players_count = len(players)
            while True:
                if await request.is_disconnected():
                    break
                # heartbeat every 1s with the current timer state
                if time.time() - last_heartbeat >= 1.0:
                    last_heartbeat = time.time()
                    # refresh the match row to get latest server state
                    ms2 = await _sb_select("onevone_matches", f"room_id=eq.{room_id}&select=*&limit=1")
                    pl2 = await _sb_select("onevone_players", f"room_id=eq.{room_id}&select=*&order=slot.asc")
                    if ms2:
                        m = ms2[0]
                        if m["state"] != last_match_state:
                            last_match_state = m["state"]
                            await _broadcast_room_event(room_id, {"type": "state", "match": _serialize_match(m)})
                    if len(pl2) != last_players_count:
                        last_players_count = len(pl2)
                        await _broadcast_room_event(room_id, {"type": "players", "players": pl2})
                    heartbeat = {
                        "type": "tick",
                        "now": _iso(_now()),
                        "match": _serialize_match(ms2[0]) if ms2 else None,
                    }
                    yield f"data: {json.dumps(heartbeat, default=str, ensure_ascii=False)}\n\n"
                # drain events
                try:
                    evt = await asyncio.wait_for(queue.get(), timeout=0.5)
                    yield f"data: {json.dumps(evt, default=str, ensure_ascii=False)}\n\n"
                except asyncio.TimeoutError:
                    pass
        finally:
            subs = _room_subs.get(room_id, [])
            if queue in subs:
                subs.remove(queue)

    return StreamingResponse(event_gen(), media_type="text/event-stream",
                             headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})


# --------------------------------------------------------------------------- #
#  Server-side timer task — drives the state machine
# --------------------------------------------------------------------------- #

async def _run_match_timer(match_id: str, room_id: str, main_dur: int, overtime_dur: int) -> None:
    """Advance the match:
        countdown(3s) -> ready (wait for both clients, max READY_TIMEOUT_S)
                       -> playing(main_dur) -> overtime(overtime_dur) -> finished
    The 'ready' phase ensures neither client can submit/win before the
    other has loaded the challenge. If only one client signals ready in
    time, that client wins (reason='opponent_no_show')."""
    from datetime import timedelta
    try:
        # 1) 3 s countdown
        await asyncio.sleep(3)
        # 2) Transition to 'ready' — wait for both clients to load + /ready
        await _sb_update("onevone_matches", f"id=eq.{match_id}", {"state": "ready"})
        await _broadcast_room_event(room_id, {"type": "state"})

        ps = await _sb_select("onevone_players", f"room_id=eq.{room_id}&select=user_id")
        expected_ready = {str(p["user_id"]) for p in ps}
        ready_set = _match_ready.setdefault(match_id, set())
        ready_signal = _match_ready_signal.setdefault(match_id, asyncio.Event())
        deadline = _now() + timedelta(seconds=READY_TIMEOUT_S)
        while True:
            if expected_ready.issubset(ready_set):
                break
            remaining = (deadline - _now()).total_seconds()
            if remaining <= 0:
                break
            try:
                await asyncio.wait_for(ready_signal.wait(), timeout=min(remaining, 2.0))
            except asyncio.TimeoutError:
                pass
            ready_signal.clear()

        if not expected_ready.issubset(ready_set):
            missing = expected_ready - ready_set
            present = expected_ready & ready_set
            winner = next(iter(present)) if len(present) == 1 else None
            reason = "opponent_no_show" if winner else "abandoned"
            await _sb_update("onevone_matches", f"id=eq.{match_id}", {
                "state": "finished",
                "winner_user_id": winner,
                "win_reason": reason,
            })
            await _broadcast_room_event(room_id, {
                "type": "match_finished",
                "winner": winner,
                "reason": reason,
            })
            return

        # 3) Both ready — start the actual match
        playing_started_at = _now()
        await _sb_update("onevone_matches", f"id=eq.{match_id}", {
            "state": "playing",
            "started_at": _iso(playing_started_at),
            "ends_at": _iso(playing_started_at + timedelta(seconds=main_dur)),
        })
        await _broadcast_room_event(room_id, {"type": "state"})
        await asyncio.sleep(main_dur)
        # check if already finished (winner)
        m = await _sb_select("onevone_matches", f"id=eq.{match_id}&select=state,winner_user_id&limit=1")
        if not m or m[0]["state"] == "finished":
            return
        # enter overtime
        await _sb_update("onevone_matches", f"id=eq.{match_id}", {
            "state": "overtime",
            "overtime_ends_at": _iso(_now() + timedelta(seconds=overtime_dur)),
        })
        await _broadcast_room_event(room_id, {"type": "state"})
        await asyncio.sleep(overtime_dur)
        # check again
        m = await _sb_select("onevone_matches", f"id=eq.{match_id}&select=state,winner_user_id&limit=1")
        if not m or m[0]["state"] == "finished":
            return
        # draw
        await _sb_update("onevone_matches", f"id=eq.{match_id}", {
            "state": "finished",
            "winner_user_id": None,
            "win_reason": "overtime_draw",
        })
        await _broadcast_room_event(room_id, {
            "type": "match_finished",
            "winner": None,
            "reason": "overtime_draw",
        })
    except asyncio.CancelledError:
        pass
    except Exception as e:
        print(f"[onevone] timer error for match {match_id}: {e}")
    finally:
        _match_ready.pop(match_id, None)
        _match_ready_signal.pop(match_id, None)


def _relativedelta(seconds: int):
    """Tiny shim so we don't need to import datetime.timedelta everywhere."""
    from datetime import timedelta
    return timedelta(seconds=seconds)