feat: complete realtime challenge matchmaking

2026-08-09 03:07:38 +08:00
parent b6fc2431ca
commit 821311f4ad
11 changed files with 767 additions and 31 deletions
@@ -62,11 +62,32 @@ class QuestionVersionAdmin(admin.ModelAdmin):


 admin.site.register(ContestAnswer)
-admin.site.register(RealtimeMatch)
 admin.site.register(RatingHistory)
 admin.site.register(LeaderboardSnapshot)


+@admin.register(RealtimeMatch)
+class RealtimeMatchAdmin(admin.ModelAdmin):
+    list_display = (
+        "id",
+        "contest",
+        "match_type",
+        "challenge_code",
+        "player_one",
+        "player_two",
+        "status",
+        "created_at",
+    )
+    list_filter = ("match_type", "status", "contest__track")
+    search_fields = (
+        "challenge_code",
+        "player_one__username",
+        "player_two__username",
+    )
+    readonly_fields = ("created_at", "started_at", "completed_at")
+    ordering = ("-created_at",)
+
+
@admin.register(MathGameAttempt)
 class MathGameAttemptAdmin(admin.ModelAdmin):
    list_display = (
@@ -26,7 +26,11 @@ class MatchConsumer(AsyncJsonWebsocketConsumer):
            await self.send_json({"type": "pong"})
            return
        if event_type == "progress":
-            answered_count = max(0, min(int(content.get("answered_count", 0)), 100))
+            try:
+                answered_count = max(0, min(int(content.get("answered_count", 0)), 100))
+            except (TypeError, ValueError):
+                await self.send_json({"type": "error", "message": "答题进度无效"})
+                return
            await self.channel_layer.group_send(
                self.group_name,
                {
@@ -45,6 +49,15 @@ class MatchConsumer(AsyncJsonWebsocketConsumer):
            }
        )

+    async def match_state(self, event):
+        await self.send_json(
+            {
+                "type": "state",
+                "reason": event["reason"],
+                "match_id": str(self.match_id),
+            }
+        )
+
    @database_sync_to_async
    def _is_participant(self, user_id):
        return RealtimeMatch.objects.filter(id=self.match_id).filter(
@@ -0,0 +1,36 @@
+# Generated by Django 4.2.23 on 2026-08-08 18:41
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('contest', '0003_mathgameattempt_and_more'),
+    ]
+
+    operations = [
+        migrations.RemoveIndex(
+            model_name='realtimematch',
+            name='matchmaking_lookup_idx',
+        ),
+        migrations.AddField(
+            model_name='realtimematch',
+            name='challenge_code',
+            field=models.CharField(blank=True, max_length=8, null=True, unique=True),
+        ),
+        migrations.AddField(
+            model_name='realtimematch',
+            name='expires_at',
+            field=models.DateTimeField(blank=True, null=True),
+        ),
+        migrations.AddField(
+            model_name='realtimematch',
+            name='match_type',
+            field=models.CharField(choices=[('random', '随机匹配'), ('challenge', '联机码约战')], default='random', max_length=16),
+        ),
+        migrations.AddIndex(
+            model_name='realtimematch',
+            index=models.Index(fields=['contest', 'match_type', 'status', 'player_one_rating', 'created_at'], name='matchmaking_lookup_idx'),
+        ),
+    ]
@@ -77,6 +77,10 @@ class ContestQuestion(models.Model):


 class RealtimeMatch(models.Model):
+    class MatchType(models.TextChoices):
+        RANDOM = "random", "随机匹配"
+        CHALLENGE = "challenge", "联机码约战"
+
    class Status(models.TextChoices):
        WAITING = "waiting", "等待对手"
        ACTIVE = "active", "进行中"
@@ -85,6 +89,12 @@ class RealtimeMatch(models.Model):

    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
    contest = models.ForeignKey(Contest, on_delete=models.PROTECT)
+    match_type = models.CharField(
+        max_length=16,
+        choices=MatchType.choices,
+        default=MatchType.RANDOM,
+    )
+    challenge_code = models.CharField(max_length=8, unique=True, null=True, blank=True)
    player_one = models.ForeignKey(
        settings.AUTH_USER_MODEL, on_delete=models.PROTECT, related_name="matches_as_player_one"
    )
@@ -106,13 +116,20 @@ class RealtimeMatch(models.Model):
        related_name="won_matches",
    )
    created_at = models.DateTimeField(auto_now_add=True)
+    expires_at = models.DateTimeField(null=True, blank=True)
    started_at = models.DateTimeField(null=True, blank=True)
    completed_at = models.DateTimeField(null=True, blank=True)

    class Meta:
        indexes = [
            models.Index(
-                fields=("contest", "status", "player_one_rating", "created_at"),
+                fields=(
+                    "contest",
+                    "match_type",
+                    "status",
+                    "player_one_rating",
+                    "created_at",
+                ),
                name="matchmaking_lookup_idx",
            )
        ]
@@ -1,5 +1,9 @@
+import secrets
+from datetime import timedelta
 from decimal import Decimal

+from asgiref.sync import async_to_sync
+from channels.layers import get_channel_layer
 from django.db import transaction
 from django.db.models import Q
 from django.utils import timezone
@@ -16,6 +20,93 @@ from .models import (
    RealtimeMatch,
 )

+CHALLENGE_CODE_ALPHABET = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789"
+WAITING_MATCH_TTL = timedelta(minutes=10)
+
+
+def _broadcast_match(match_id, reason):
+    channel_layer = get_channel_layer()
+    if channel_layer is None:
+        return
+    async_to_sync(channel_layer.group_send)(
+        f"match_{match_id}",
+        {
+            "type": "match.state",
+            "reason": reason,
+        },
+    )
+
+
+def notify_match_on_commit(match_id, reason):
+    transaction.on_commit(lambda: _broadcast_match(match_id, reason))
+
+
+def _new_challenge_code():
+    for _ in range(20):
+        code = "".join(secrets.choice(CHALLENGE_CODE_ALPHABET) for _ in range(6))
+        if not RealtimeMatch.objects.filter(challenge_code=code).exists():
+            return code
+    raise ValidationError("暂时无法生成联机码，请稍后重试")
+
+
+def _validate_realtime_contest(contest):
+    if contest.kind != Contest.Kind.REALTIME or contest.status != Contest.Status.PUBLISHED:
+        raise ValidationError("实时比赛不可用")
+
+
+def _cancel_expired_waiting_matches():
+    RealtimeMatch.objects.filter(status=RealtimeMatch.Status.WAITING).filter(
+        Q(expires_at__isnull=True) | Q(expires_at__lte=timezone.now())
+    ).update(status=RealtimeMatch.Status.CANCELLED)
+
+
+def _active_match_for(user):
+    return (
+        RealtimeMatch.objects.filter(
+            Q(player_one=user) | Q(player_two=user),
+            status=RealtimeMatch.Status.ACTIVE,
+        )
+        .select_related("contest", "player_one", "player_two")
+        .first()
+    )
+
+
+def _cancel_other_waiting_matches(user, match_type):
+    matches = list(
+        RealtimeMatch.objects.filter(
+            player_one=user,
+            status=RealtimeMatch.Status.WAITING,
+        )
+        .exclude(match_type=match_type)
+        .values_list("id", flat=True)
+    )
+    if matches:
+        RealtimeMatch.objects.filter(id__in=matches).update(
+            status=RealtimeMatch.Status.CANCELLED
+        )
+        for match_id in matches:
+            notify_match_on_commit(match_id, "cancelled")
+
+
+def _activate_match(match, user):
+    now = timezone.now()
+    match.player_two = user
+    match.player_two_rating = user.rating
+    match.status = RealtimeMatch.Status.ACTIVE
+    match.started_at = now
+    match.save(
+        update_fields=["player_two", "player_two_rating", "status", "started_at"]
+    )
+    ContestAttempt.objects.bulk_create(
+        [
+            ContestAttempt(contest=match.contest, user=match.player_one, match=match),
+            ContestAttempt(contest=match.contest, user=user, match=match),
+        ]
+    )
+    match.attempts.update(started_at=now)
+    notify_match_on_commit(match.id, "matched")
+    return match
+

 def normalize_answer(value):
    text = str(value).strip().lower().replace(" ", "")
@@ -35,12 +126,12 @@ def attempt_payload(attempt, include_results=False):
            "metadata": item.question_version.metadata,
            "points": item.points,
        }
-        if include_results and item.id in answers:
-            answer = answers[item.id]
+        if include_results:
+            answer = answers.get(item.id)
            question.update(
                {
-                    "submitted_answer": answer.submitted_answer,
-                    "is_correct": answer.is_correct,
+                    "submitted_answer": answer.submitted_answer if answer else "",
+                    "is_correct": answer.is_correct if answer else False,
                    "correct_answer": item.question_version.answer,
                    "explanation": item.question_version.explanation,
                }
@@ -48,6 +139,7 @@ def attempt_payload(attempt, include_results=False):
        questions.append(question)
    return {
        "attempt_id": attempt.id,
+        "match_id": attempt.match_id,
        "contest": attempt.contest.title,
        "kind": attempt.contest.kind,
        "status": attempt.status,
@@ -90,7 +182,10 @@ def submit_attempt(user, attempt_id, raw_answers, submission_key):
    )
    if attempt.status != ContestAttempt.Status.ACTIVE:
        if submission_key and attempt.submission_key == submission_key:
-            return attempt_payload(attempt, include_results=True)
+            include_results = not attempt.match_id or (
+                attempt.match.status == RealtimeMatch.Status.COMPLETED
+            )
+            return attempt_payload(attempt, include_results=include_results)
        raise ValidationError("该答题记录已经结算")
    if not submission_key:
        raise ValidationError({"Idempotency-Key": "正式提交必须提供幂等键"})
@@ -153,20 +248,34 @@ def submit_attempt(user, attempt_id, raw_answers, submission_key):
            },
        )
    if attempt.match_id:
-        finalize_match(attempt.match_id)
+        match = finalize_match(attempt.match_id)
+        if match.status != RealtimeMatch.Status.COMPLETED:
+            notify_match_on_commit(attempt.match_id, "submitted")
+        attempt.match.refresh_from_db()
+        return attempt_payload(
+            attempt,
+            include_results=attempt.match.status == RealtimeMatch.Status.COMPLETED,
+        )
    return attempt_payload(attempt, include_results=True)


@transaction.atomic
 def find_match(user, contest):
-    if contest.kind != Contest.Kind.REALTIME or contest.status != Contest.Status.PUBLISHED:
-        raise ValidationError("实时比赛不可用")
+    _validate_realtime_contest(contest)
+    _cancel_expired_waiting_matches()
+    active = _active_match_for(user)
+    if active:
+        if active.contest_id == contest.id:
+            return active
+        raise ValidationError("你已有一场进行中的实时比赛")
+    _cancel_other_waiting_matches(user, RealtimeMatch.MatchType.RANDOM)
    existing = (
        RealtimeMatch.objects.filter(
            Q(player_one=user) | Q(player_two=user),
            contest=contest,
+            match_type=RealtimeMatch.MatchType.RANDOM,
            status__in=[RealtimeMatch.Status.WAITING, RealtimeMatch.Status.ACTIVE],
-        )
+        ).filter(Q(status=RealtimeMatch.Status.ACTIVE) | Q(expires_at__gt=timezone.now()))
        .order_by("-created_at")
        .first()
    )
@@ -177,7 +286,9 @@ def find_match(user, contest):
        RealtimeMatch.objects.select_for_update(skip_locked=True)
        .filter(
            contest=contest,
+            match_type=RealtimeMatch.MatchType.RANDOM,
            status=RealtimeMatch.Status.WAITING,
+            expires_at__gt=timezone.now(),
            player_one_rating__gte=max(0, user.rating - 300),
            player_one_rating__lte=user.rating + 300,
        )
@@ -188,42 +299,198 @@ def find_match(user, contest):
    if waiting is None:
        return RealtimeMatch.objects.create(
            contest=contest,
+            match_type=RealtimeMatch.MatchType.RANDOM,
            player_one=user,
            player_one_rating=user.rating,
+            expires_at=timezone.now() + WAITING_MATCH_TTL,
        )

-    waiting.player_two = user
-    waiting.player_two_rating = user.rating
-    waiting.status = RealtimeMatch.Status.ACTIVE
-    waiting.started_at = timezone.now()
-    waiting.save(
-        update_fields=["player_two", "player_two_rating", "status", "started_at"]
+    return _activate_match(waiting, user)
+
+
+@transaction.atomic
+def create_challenge(user, contest):
+    _validate_realtime_contest(contest)
+    _cancel_expired_waiting_matches()
+    active = _active_match_for(user)
+    if active:
+        raise ValidationError("你已有一场进行中的实时比赛")
+    _cancel_other_waiting_matches(user, RealtimeMatch.MatchType.CHALLENGE)
+    existing = (
+        RealtimeMatch.objects.filter(
+            player_one=user,
+            contest=contest,
+            match_type=RealtimeMatch.MatchType.CHALLENGE,
+            status=RealtimeMatch.Status.WAITING,
+            expires_at__gt=timezone.now(),
+        )
+        .order_by("-created_at")
+        .first()
    )
-    ContestAttempt.objects.bulk_create(
-        [
-            ContestAttempt(contest=contest, user=waiting.player_one, match=waiting),
-            ContestAttempt(contest=contest, user=user, match=waiting),
-        ]
+    if existing:
+        return existing
+    return RealtimeMatch.objects.create(
+        contest=contest,
+        match_type=RealtimeMatch.MatchType.CHALLENGE,
+        challenge_code=_new_challenge_code(),
+        player_one=user,
+        player_one_rating=user.rating,
+        expires_at=timezone.now() + WAITING_MATCH_TTL,
    )
-    return waiting
+
+
+@transaction.atomic
+def join_challenge(user, challenge_code):
+    _cancel_expired_waiting_matches()
+    code = str(challenge_code or "").strip().upper()
+    if len(code) != 6 or any(character not in CHALLENGE_CODE_ALPHABET for character in code):
+        raise ValidationError({"challenge_code": "联机码应为 6 位大写字母或数字"})
+    try:
+        match = (
+            RealtimeMatch.objects.select_for_update()
+            .select_related("contest", "player_one")
+            .get(
+                challenge_code=code,
+                match_type=RealtimeMatch.MatchType.CHALLENGE,
+            )
+        )
+    except RealtimeMatch.DoesNotExist as exc:
+        raise ValidationError({"challenge_code": "联机码不存在"}) from exc
+    if match.player_one_id == user.id:
+        raise ValidationError({"challenge_code": "不能加入自己创建的约战"})
+    if match.status != RealtimeMatch.Status.WAITING or (
+        match.expires_at and match.expires_at <= timezone.now()
+    ):
+        raise ValidationError({"challenge_code": "联机码已失效或已被使用"})
+    active = _active_match_for(user)
+    if active and active.id != match.id:
+        raise ValidationError("你已有一场进行中的实时比赛")
+    own_waiting_ids = list(
+        RealtimeMatch.objects.filter(
+            player_one=user,
+            status=RealtimeMatch.Status.WAITING,
+        )
+        .exclude(id=match.id)
+        .values_list("id", flat=True)
+    )
+    if own_waiting_ids:
+        RealtimeMatch.objects.filter(id__in=own_waiting_ids).update(
+            status=RealtimeMatch.Status.CANCELLED
+        )
+        for match_id in own_waiting_ids:
+            notify_match_on_commit(match_id, "cancelled")
+    return _activate_match(match, user)
+
+
+@transaction.atomic
+def cancel_waiting_match(user, match_id):
+    match = RealtimeMatch.objects.select_for_update().get(id=match_id)
+    if match.player_one_id != user.id:
+        raise ValidationError("只有创建者可以取消等待")
+    if match.status != RealtimeMatch.Status.WAITING:
+        raise ValidationError("只能取消等待中的比赛")
+    match.status = RealtimeMatch.Status.CANCELLED
+    match.save(update_fields=["status"])
+    notify_match_on_commit(match.id, "cancelled")
+    return match


 def match_payload(match, user):
-    attempt = match.attempts.filter(user=user).first()
+    reveal_results = match.status == RealtimeMatch.Status.COMPLETED
+    attempts = {
+        attempt.user_id: attempt
+        for attempt in match.attempts.select_related("user", "contest").all()
+    }
+    attempt = attempts.get(user.id)
    opponent = match.player_two if match.player_one_id == user.id else match.player_one
+    opponent_attempt = attempts.get(opponent.id) if opponent else None
+    rating_change = (
+        match.rating_changes.filter(user=user).values("delta", "rating_after").first()
+        if reveal_results
+        else None
+    )
    return {
        "match_id": match.id,
+        "match_type": match.match_type,
+        "is_owner": match.player_one_id == user.id,
+        "challenge_code": (
+            match.challenge_code
+            if match.match_type == RealtimeMatch.MatchType.CHALLENGE
+            and match.status == RealtimeMatch.Status.WAITING
+            else None
+        ),
        "status": match.status,
+        "contest": match.contest.title,
+        "duration_seconds": match.contest.duration_seconds,
+        "expires_at": match.expires_at,
+        "started_at": match.started_at,
        "opponent": (
-            {"nickname": opponent.nickname, "rating": opponent.rating}
+            {
+                "nickname": opponent.nickname,
+                "rating": opponent.rating,
+                "status": opponent_attempt.status if opponent_attempt else None,
+                "score": opponent_attempt.score if reveal_results and opponent_attempt else None,
+                "correct_count": (
+                    opponent_attempt.correct_count
+                    if reveal_results and opponent_attempt
+                    else None
+                ),
+            }
            if opponent
            else None
        ),
-        "attempt": attempt_payload(attempt) if attempt else None,
+        "attempt": (
+            attempt_payload(attempt, include_results=reveal_results)
+            if attempt
+            else None
+        ),
+        "result": (
+            {
+                "winner": (
+                    "draw"
+                    if match.winner_id is None
+                    else "self"
+                    if match.winner_id == user.id
+                    else "opponent"
+                ),
+                "rating_delta": rating_change["delta"] if rating_change else 0,
+                "rating_after": rating_change["rating_after"] if rating_change else user.rating,
+            }
+            if reveal_results
+            else None
+        ),
        "websocket_path": f"/ws/v1/contest/matches/{match.id}/",
    }


+@transaction.atomic
+def refresh_match_state(match_id):
+    match = (
+        RealtimeMatch.objects.select_for_update()
+        .select_related("contest", "player_one", "player_two")
+        .get(id=match_id)
+    )
+    if (
+        match.status == RealtimeMatch.Status.WAITING
+        and match.expires_at
+        and match.expires_at <= timezone.now()
+    ):
+        match.status = RealtimeMatch.Status.CANCELLED
+        match.save(update_fields=["status"])
+        notify_match_on_commit(match.id, "expired")
+    elif match.status == RealtimeMatch.Status.ACTIVE and match.started_at:
+        deadline = match.started_at + timedelta(seconds=match.contest.duration_seconds)
+        if timezone.now() >= deadline:
+            elapsed_ms = match.contest.duration_seconds * 1000
+            match.attempts.filter(status=ContestAttempt.Status.ACTIVE).update(
+                status=ContestAttempt.Status.EXPIRED,
+                duration_ms=elapsed_ms,
+                submitted_at=timezone.now(),
+            )
+            match = finalize_match(match.id)
+    return match
+
+
 def _elo_delta(rating, opponent_rating, score, k=32):
    expected = 1 / (1 + 10 ** ((opponent_rating - rating) / 400))
    return round(k * (score - expected))
@@ -277,4 +544,5 @@ def finalize_match(match_id):
    match.status = RealtimeMatch.Status.COMPLETED
    match.completed_at = timezone.now()
    match.save(update_fields=["winner", "status", "completed_at"])
+    notify_match_on_commit(match.id, "completed")
    return match
@@ -0,0 +1,87 @@
+import pytest
+from asgiref.sync import async_to_sync
+from channels.layers import get_channel_layer
+from channels.routing import URLRouter
+from channels.testing import WebsocketCommunicator
+from django.urls import path
+
+from accounts.models import User
+from contest.consumers import MatchConsumer
+from contest.models import Contest, Question, RealtimeMatch
+
+
+@pytest.mark.django_db(transaction=True)
+def test_match_consumer_双方连接并同步答题进度():
+    first = User.objects.create_user(
+        username="socket_player_one",
+        password="StrongPass_2026",
+        nickname="WS 玩家一",
+    )
+    second = User.objects.create_user(
+        username="socket_player_two",
+        password="StrongPass_2026",
+        nickname="WS 玩家二",
+    )
+    outsider = User.objects.create_user(
+        username="socket_outsider",
+        password="StrongPass_2026",
+        nickname="WS 局外人",
+    )
+    contest = Contest.objects.create(
+        slug="socket-contest",
+        title="WebSocket 联机赛",
+        kind=Contest.Kind.REALTIME,
+        track=Question.Track.STANDARD,
+        status=Contest.Status.PUBLISHED,
+    )
+    match = RealtimeMatch.objects.create(
+        contest=contest,
+        player_one=first,
+        player_two=second,
+        player_one_rating=first.rating,
+        player_two_rating=second.rating,
+        status=RealtimeMatch.Status.ACTIVE,
+    )
+    application = URLRouter(
+        [
+            path(
+                "ws/test/<uuid:match_id>/",
+                MatchConsumer.as_asgi(),
+            )
+        ]
+    )
+
+    async def scenario():
+        outsider_socket = WebsocketCommunicator(application, f"/ws/test/{match.id}/")
+        outsider_socket.scope["user"] = outsider
+        outsider_connected, close_code = await outsider_socket.connect()
+        assert not outsider_connected
+        assert close_code == 4403
+
+        first_socket = WebsocketCommunicator(application, f"/ws/test/{match.id}/")
+        second_socket = WebsocketCommunicator(application, f"/ws/test/{match.id}/")
+        first_socket.scope["user"] = first
+        second_socket.scope["user"] = second
+        first_connected, _ = await first_socket.connect()
+        second_connected, _ = await second_socket.connect()
+        assert first_connected and second_connected
+        assert (await first_socket.receive_json_from())["type"] == "connected"
+        assert (await second_socket.receive_json_from())["type"] == "connected"
+
+        await first_socket.send_json_to({"type": "progress", "answered_count": 3})
+        first_progress = await first_socket.receive_json_from()
+        second_progress = await second_socket.receive_json_from()
+        assert first_progress["answered_count"] == 3
+        assert second_progress["answered_count"] == 3
+        assert second_progress["user_id"] == str(first.id)
+
+        await get_channel_layer().group_send(
+            f"match_{match.id}",
+            {"type": "match.state", "reason": "completed"},
+        )
+        assert (await first_socket.receive_json_from())["reason"] == "completed"
+        assert (await second_socket.receive_json_from())["reason"] == "completed"
+        await first_socket.disconnect()
+        await second_socket.disconnect()
+
+    async_to_sync(scenario)()
@@ -0,0 +1,96 @@
+import pytest
+from django.test import Client
+
+from accounts.models import User
+from contest.models import Contest, ContestQuestion, Question, QuestionVersion
+
+
+@pytest.fixture
+def realtime_api_setup(db):
+    question = Question.objects.create(
+        slug="realtime-api-question",
+        track=Question.Track.STANDARD,
+    )
+    version = QuestionVersion.objects.create(
+        question=question,
+        version=1,
+        prompt="18 + 24",
+        answer="42",
+        explanation="18 + 24 = 42",
+    )
+    contest = Contest.objects.create(
+        slug="realtime-api",
+        title="API 联机赛",
+        kind=Contest.Kind.REALTIME,
+        track=Question.Track.STANDARD,
+        status=Contest.Status.PUBLISHED,
+        duration_seconds=60,
+    )
+    ContestQuestion.objects.create(
+        contest=contest,
+        question_version=version,
+        order=1,
+        points=100,
+    )
+    first = User.objects.create_user(
+        username="api_player_one",
+        password="StrongPass_2026",
+        nickname="API 玩家一",
+    )
+    second = User.objects.create_user(
+        username="api_player_two",
+        password="StrongPass_2026",
+        nickname="API 玩家二",
+    )
+    first_client = Client()
+    second_client = Client()
+    first_client.force_login(first)
+    second_client.force_login(second)
+    return contest, first_client, second_client
+
+
+@pytest.mark.django_db
+def test_challenge_api_创建加入状态提交形成完整闭环(realtime_api_setup):
+    contest, first_client, second_client = realtime_api_setup
+    created = first_client.post(
+        f"/api/v1/contests/{contest.slug}/challenges/",
+        {},
+        content_type="application/json",
+    )
+    code = created.json()["challenge_code"]
+
+    joined = second_client.post(
+        "/api/v1/contests/challenges/join/",
+        {"challenge_code": code.lower()},
+        content_type="application/json",
+    )
+    match_id = joined.json()["match_id"]
+    first_state = first_client.get(f"/api/v1/contests/matches/{match_id}/")
+
+    assert created.status_code == 201
+    assert joined.status_code == 200
+    assert joined.json()["status"] == "active"
+    assert first_state.json()["opponent"]["nickname"] == "API 玩家二"
+
+    first_attempt = first_state.json()["attempt"]["attempt_id"]
+    second_attempt = joined.json()["attempt"]["attempt_id"]
+    first_submit = first_client.post(
+        f"/api/v1/contests/attempts/{first_attempt}/submit/",
+        {"answers": [{"order": 1, "answer": "42"}]},
+        content_type="application/json",
+        HTTP_IDEMPOTENCY_KEY="api-submit-one",
+    )
+    second_submit = second_client.post(
+        f"/api/v1/contests/attempts/{second_attempt}/submit/",
+        {"answers": [{"order": 1, "answer": "0"}]},
+        content_type="application/json",
+        HTTP_IDEMPOTENCY_KEY="api-submit-two",
+    )
+
+    assert first_submit.json()["status"] == "active"
+    assert "correct_answer" not in first_submit.json()["attempt"]["questions"][0]
+    assert second_submit.json()["status"] == "completed"
+    assert second_submit.json()["result"]["winner"] == "opponent"
+    final_state = first_client.get(f"/api/v1/contests/matches/{match_id}/").json()
+    assert final_state["result"]["winner"] == "self"
+    assert final_state["attempt"]["questions"][0]["correct_answer"] == "42"
@@ -15,9 +15,14 @@ from contest.models import (
    RealtimeMatch,
 )
 from contest.services import (
+    cancel_waiting_match,
+    create_challenge,
    finalize_match,
    find_match,
+    join_challenge,
+    match_payload,
    normalize_answer,
+    refresh_match_state,
    start_attempt,
    submit_attempt,
 )
@@ -62,6 +67,16 @@ def daily_contest(db):
    return contest


+@pytest.fixture
+def realtime_contest(daily_contest):
+    daily_contest.kind = Contest.Kind.REALTIME
+    daily_contest.slug = "realtime-with-question"
+    daily_contest.title = "联机测试赛"
+    daily_contest.duration_seconds = 60
+    daily_contest.save(update_fields=["kind", "slug", "title", "duration_seconds"])
+    return daily_contest
+
+
@pytest.mark.parametrize(
    "raw, expected",
    [
@@ -201,3 +216,147 @@ def test_find_match_同分玩家配对并由_finalize_match_唯一结算_rating(
    assert first.rating == 1016
    assert second.rating == 984
    assert RatingHistory.objects.filter(match=active).count() == 2
+
+
+@pytest.mark.django_db
+def test_challenge_code_创建者和加入者通过联机码进入同一场(realtime_contest):
+    first = User.objects.create_user(
+        username="challenge_owner",
+        password="StrongPass_2026",
+        nickname="房主",
+    )
+    second = User.objects.create_user(
+        username="challenge_guest",
+        password="StrongPass_2026",
+        nickname="访客",
+    )
+
+    waiting = create_challenge(first, realtime_contest)
+    active = join_challenge(second, waiting.challenge_code.lower())
+    owner_payload = match_payload(active, first)
+    guest_payload = match_payload(active, second)
+
+    assert len(waiting.challenge_code) == 6
+    assert active.id == waiting.id
+    assert active.match_type == RealtimeMatch.MatchType.CHALLENGE
+    assert active.status == RealtimeMatch.Status.ACTIVE
+    assert active.attempts.count() == 2
+    assert owner_payload["opponent"]["nickname"] == "访客"
+    assert guest_payload["opponent"]["nickname"] == "房主"
+    assert owner_payload["attempt"]["questions"] == guest_payload["attempt"]["questions"]
+
+
+@pytest.mark.django_db
+def test_random_match_不会加入联机码约战(realtime_contest):
+    owner = User.objects.create_user(
+        username="private_owner",
+        password="StrongPass_2026",
+        nickname="约战房主",
+    )
+    random_player = User.objects.create_user(
+        username="random_player",
+        password="StrongPass_2026",
+        nickname="随机玩家",
+    )
+
+    challenge = create_challenge(owner, realtime_contest)
+    random_match = find_match(random_player, realtime_contest)
+
+    assert challenge.status == RealtimeMatch.Status.WAITING
+    assert random_match.id != challenge.id
+    assert random_match.match_type == RealtimeMatch.MatchType.RANDOM
+
+
+@pytest.mark.django_db
+def test_realtime_submit_双方结束前不泄露答案且结束后结算(realtime_contest):
+    first = User.objects.create_user(
+        username="fair_player_one",
+        password="StrongPass_2026",
+        nickname="公平玩家一",
+    )
+    second = User.objects.create_user(
+        username="fair_player_two",
+        password="StrongPass_2026",
+        nickname="公平玩家二",
+    )
+    match = join_challenge(
+        second,
+        create_challenge(first, realtime_contest).challenge_code,
+    )
+    first_attempt = match.attempts.get(user=first)
+    second_attempt = match.attempts.get(user=second)
+
+    first_result = submit_attempt(
+        first,
+        first_attempt.id,
+        [{"order": 1, "answer": "42"}],
+        "fair-submit-one",
+    )
+    active_payload = match_payload(match, first)
+
+    assert "correct_answer" not in first_result["questions"][0]
+    assert active_payload["status"] == RealtimeMatch.Status.ACTIVE
+    assert active_payload["attempt"]["status"] == ContestAttempt.Status.SUBMITTED
+
+    submit_attempt(
+        second,
+        second_attempt.id,
+        [{"order": 1, "answer": "0"}],
+        "fair-submit-two",
+    )
+    match.refresh_from_db()
+    completed_payload = match_payload(match, first)
+
+    assert match.status == RealtimeMatch.Status.COMPLETED
+    assert completed_payload["result"]["winner"] == "self"
+    assert completed_payload["attempt"]["questions"][0]["correct_answer"] == "42"
+    assert completed_payload["opponent"]["score"] == 0
+
+
+@pytest.mark.django_db
+def test_realtime_timeout_未提交玩家自动过期并完成比赛(realtime_contest):
+    first = User.objects.create_user(
+        username="timeout_one",
+        password="StrongPass_2026",
+        nickname="超时玩家一",
+    )
+    second = User.objects.create_user(
+        username="timeout_two",
+        password="StrongPass_2026",
+        nickname="超时玩家二",
+    )
+    match = join_challenge(
+        second,
+        create_challenge(first, realtime_contest).challenge_code,
+    )
+    RealtimeMatch.objects.filter(id=match.id).update(
+        started_at=timezone.now() - timedelta(seconds=61)
+    )
+
+    refreshed = refresh_match_state(match.id)
+
+    assert refreshed.status == RealtimeMatch.Status.COMPLETED
+    assert not refreshed.attempts.filter(status=ContestAttempt.Status.ACTIVE).exists()
+
+
+@pytest.mark.django_db
+def test_challenge_owner_可取消等待中的联机码(realtime_contest):
+    owner = User.objects.create_user(
+        username="cancel_owner",
+        password="StrongPass_2026",
+        nickname="取消房主",
+    )
+    waiting = create_challenge(owner, realtime_contest)
+
+    cancelled = cancel_waiting_match(owner, waiting.id)
+
+    assert cancelled.status == RealtimeMatch.Status.CANCELLED
+    with pytest.raises(ValidationError, match="失效"):
+        join_challenge(
+            User.objects.create_user(
+                username="late_guest",
+                password="StrongPass_2026",
+                nickname="迟到访客",
+            ),
+            waiting.challenge_code,
+        )
@@ -3,8 +3,11 @@ from django.urls import path
 from .views import (
    AttemptStartView,
    AttemptSubmitView,
+    ChallengeCreateView,
+    ChallengeJoinView,
    ContestListView,
    LeaderboardView,
+    MatchCancelView,
    MatchmakingView,
    MatchStateView,
    MathGameCatalogView,
@@ -16,11 +19,18 @@ from .views import (

 urlpatterns = [
    path("", ContestListView.as_view(), name="contest-list"),
+    path("challenges/join/", ChallengeJoinView.as_view(), name="challenge-join"),
+    path("matches/<uuid:match_id>/", MatchStateView.as_view(), name="match-state"),
+    path("matches/<uuid:match_id>/cancel/", MatchCancelView.as_view(), name="match-cancel"),
    path("<slug:slug>/start/", AttemptStartView.as_view(), name="attempt-start"),
    path("<slug:slug>/matchmaking/", MatchmakingView.as_view(), name="matchmaking"),
+    path(
+        "<slug:slug>/challenges/",
+        ChallengeCreateView.as_view(),
+        name="challenge-create",
+    ),
    path("<slug:slug>/leaderboard/", LeaderboardView.as_view(), name="leaderboard"),
    path("attempts/<uuid:attempt_id>/submit/", AttemptSubmitView.as_view(), name="attempt-submit"),
-    path("matches/<uuid:match_id>/", MatchStateView.as_view(), name="match-state"),
    path("games/", MathGameCatalogView.as_view(), name="math-game-catalog"),
    path("games/history/", MathGameHistoryView.as_view(), name="math-game-history"),
    path("games/<str:kind>/start/", MathGameStartView.as_view(), name="math-game-start"),
@@ -6,8 +6,12 @@ from rest_framework.views import APIView
 from .game_services import game_payload, request_sudoku_hint, start_game, submit_game
 from .models import Contest, ContestAttempt, MathGameAttempt, RealtimeMatch
 from .services import (
+    cancel_waiting_match,
+    create_challenge,
    find_match,
+    join_challenge,
    match_payload,
+    refresh_match_state,
    start_attempt,
    submit_attempt,
 )
@@ -44,13 +48,16 @@ class AttemptStartView(APIView):

 class AttemptSubmitView(APIView):
    def post(self, request, attempt_id):
-        get_object_or_404(ContestAttempt, id=attempt_id, user=request.user)
+        attempt = get_object_or_404(ContestAttempt, id=attempt_id, user=request.user)
        payload = submit_attempt(
            user=request.user,
            attempt_id=attempt_id,
            raw_answers=request.data.get("answers", []),
            submission_key=request.headers.get("Idempotency-Key"),
        )
+        if attempt.match_id:
+            match = refresh_match_state(attempt.match_id)
+            return Response(match_payload(match, request.user))
        return Response(payload)


@@ -63,12 +70,33 @@ class MatchmakingView(APIView):

 class MatchStateView(APIView):
    def get(self, request, match_id):
-        match = get_object_or_404(
+        existing = get_object_or_404(
            RealtimeMatch.objects.select_related("player_one", "player_two"),
            id=match_id,
        )
-        if request.user.id not in (match.player_one_id, match.player_two_id):
+        if request.user.id not in (existing.player_one_id, existing.player_two_id):
            return Response(status=status.HTTP_403_FORBIDDEN)
+        match = refresh_match_state(match_id)
+        return Response(match_payload(match, request.user))
+
+
+class ChallengeCreateView(APIView):
+    def post(self, request, slug):
+        contest = get_object_or_404(Contest, slug=slug)
+        match = create_challenge(request.user, contest)
+        return Response(match_payload(match, request.user), status=status.HTTP_201_CREATED)
+
+
+class ChallengeJoinView(APIView):
+    def post(self, request):
+        match = join_challenge(request.user, request.data.get("challenge_code"))
+        return Response(match_payload(match, request.user))
+
+
+class MatchCancelView(APIView):
+    def post(self, request, match_id):
+        get_object_or_404(RealtimeMatch, id=match_id)
+        match = cancel_waiting_match(request.user, match_id)
        return Response(match_payload(match, request.user))