wpp/weirsoe-party-protocol
8
0
Fork 0
Code Issues 19 Pull Requests Actions Packages Projects Releases Wiki Activity
Files
251ccfce19c127718fd1a86e7be4bd365908738d
weirsoe-party-protocol/docs/plans/2026-03-09-fupogfakta-implementation-plan.md
Asger Geel Weirsøe d15abf9d78
All checks were successful
CI / test-and-quality (pull_request) Successful in 2m46s
Details
CI / test-and-quality (push) Successful in 2m50s
Details
docs: add fupogfakta game engine implementation plan 
15 tasks across 8 batches covering:
- Celery infrastructure
- GameRun model + GameDriver interface
- FupOgFaktaConfig relational presets
- LieReaction model, reveal_order, ScoreEvent removal
- Full FupOgFaktaDriver with all phase transitions
- Platform play/pause/exit endpoints
- Fupogfakta lie/guess/react endpoints
- Angular frontend game screens rebuild
- Cleanup of obsolete manual-advance endpoints

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-03-09 07:38:04 +01:00

2181 lines
72 KiB
Markdown
Raw Blame History

# Fup og Fakta — Game Engine Implementation Plan
> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.
**Goal:** Implement a fully working Fup og Fakta game on top of a pluggable game-platform engine with Celery-driven timers, incremental reveal scoring, emoji reactions, and post-game awards.
**Architecture:** Platform layer (`lobby/`) provides `GameRun`, `GameDriver` interface, and Celery timer dispatch. Each game cartridge (starting with `fupogfakta/`) owns its models, config, and phase logic. The Angular frontend gets proper game screens replacing the current developer control panels.
**Tech Stack:** Django 6.0.2, Celery + Redis, Django Channels (WebSocket already built), Angular 19 standalone components, Vitest.
---
## Reading Before You Start
- `docs/plans/2026-03-09-fupogfakta-game-engine-design.md` — full design, read this first
- `lobby/views.py` — existing REST endpoints (most will be replaced)
- `fupogfakta/models.py` — existing models
- `realtime/broadcast.py` — sync_broadcast_phase_event helper
- `partyhub/settings.py` — env-driven config pattern used throughout
**Run tests before starting:** `.venv/bin/python manage.py test lobby realtime` → must be 78/78 green.
---
## Batch 1 — Celery Infrastructure
No breaking changes. Additive only. Existing tests stay green throughout.
---
### Task 1: Add Celery to dependencies
**Files:**
- Modify: `requirements.txt`
**Step 1: Add celery and redis to requirements**
```
celery>=5.3,<6
redis>=5.0,<6
```
`redis` is the pure-Python Redis client Celery uses as a broker. (`channels-redis` uses it too but declares it as a dep — add explicitly for clarity.)
**Step 2: Install**
```bash
.venv/bin/pip install celery>=5.3,<6 redis>=5.0,<6
```
Expected: installs without errors.
**Step 3: Commit**
```bash
git add requirements.txt
git commit -m "chore: add celery + redis to requirements"
```
---
### Task 2: Create Celery app
**Files:**
- Create: `partyhub/celery.py`
- Modify: `partyhub/__init__.py`
**Step 1: Create `partyhub/celery.py`**
```python
import os
from celery import Celery
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'partyhub.settings')
app = Celery('partyhub')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()
```
**Step 2: Read `partyhub/__init__.py` first, then update it**
Add at the top (after any existing content):
```python
from .celery import app as celery_app
__all__ = ('celery_app',)
```
**Step 3: Add Celery config to `partyhub/settings.py`**
Add after the `CHANNEL_LAYERS` block:
```python
CELERY_BROKER_URL = f"redis://{env('CELERY_REDIS_HOST', '127.0.0.1')}:{env('CELERY_REDIS_PORT', '6379')}/1"
CELERY_RESULT_BACKEND = CELERY_BROKER_URL
CELERY_TASK_SERIALIZER = 'json'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TIMEZONE = TIME_ZONE
# In tests: use synchronous task execution (no broker needed)
CELERY_TASK_ALWAYS_EAGER = _testing
CELERY_TASK_EAGER_PROPAGATES = _testing
```
**Step 4: Verify Django check still passes**
```bash
.venv/bin/python manage.py check
```
Expected: `System check identified no issues (0 silenced).`
**Step 5: Commit**
```bash
git add partyhub/celery.py partyhub/__init__.py partyhub/settings.py
git commit -m "feat(platform): add celery app and broker config"
```
---
### Task 3: Create the timer Celery task (skeleton)
**Files:**
- Create: `lobby/tasks.py`
**Step 1: Write failing test**
In `lobby/tests.py`, add at the bottom:
```python
class TimerTaskStaleGuardTest(TestCase):
def test_stale_task_does_nothing_when_run_not_found(self):
from lobby.tasks import handle_timer_expired
# Should not raise — run_id that does not exist is silently ignored
handle_timer_expired(run_id=99999, expected_state="LIE_PHASE")
```
**Step 2: Run test to verify it fails**
```bash
.venv/bin/python manage.py test lobby.tests.TimerTaskStaleGuardTest --verbosity=2
```
Expected: `ImportError` or `ModuleNotFoundError` for `lobby.tasks`.
**Step 3: Create `lobby/tasks.py`**
```python
import logging
from celery import shared_task
logger = logging.getLogger(__name__)
@shared_task(bind=True, max_retries=0)
def handle_timer_expired(self, run_id: int, expected_state: str) -> None:
"""
Fired by Celery when a phase timer expires.
Guards against stale tasks: if the GameRun no longer exists or its
current_state no longer matches expected_state (e.g. after pause/resume),
the task silently exits.
Full driver dispatch is wired in Batch 5 once GameRun model exists.
"""
# Import here to avoid circular imports at module load time
try:
from lobby.models import GameRun # will exist after Batch 2
except ImportError:
logger.warning("handle_timer_expired: GameRun model not yet available (pre-migration)")
return
try:
run = GameRun.objects.select_related('session').get(pk=run_id)
except GameRun.DoesNotExist:
logger.info("handle_timer_expired: run %s not found — stale task, ignoring", run_id)
return
if run.current_state != expected_state:
logger.info(
"handle_timer_expired: run %s state is %r, expected %r — stale task, ignoring",
run_id, run.current_state, expected_state,
)
return
# Full transition logic wired in Batch 5
logger.info("handle_timer_expired: run %s state=%r — TODO: dispatch driver", run_id, expected_state)
```
**Step 4: Run test to verify it passes**
```bash
.venv/bin/python manage.py test lobby.tests.TimerTaskStaleGuardTest --verbosity=2
```
Expected: PASS (ImportError path is handled gracefully).
**Step 5: Commit**
```bash
git add lobby/tasks.py lobby/tests.py
git commit -m "feat(platform): add handle_timer_expired celery task skeleton"
```
---
## Batch 2 — GameRun Model + GameDriver Interface
---
### Task 4: Add `game_type` field to GameSession and simplify status
**Files:**
- Modify: `fupogfakta/models.py`
- Create: `fupogfakta/migrations/0005_gamesession_game_type_status_active.py` (auto-generated)
**Step 1: Write failing test**
In `lobby/tests.py`, in an appropriate test class or new one:
```python
class GameSessionGameTypeTest(TestCase):
def setUp(self):
self.user = User.objects.create_user(username='host2', password='pw')
def test_game_session_has_game_type_field(self):
session = GameSession.objects.create(host=self.user, code='GGTYPE')
self.assertEqual(session.game_type, 'fupogfakta')
def test_game_session_status_active_exists(self):
session = GameSession.objects.create(host=self.user, code='ACTSTS')
session.status = GameSession.Status.ACTIVE
session.save()
session.refresh_from_db()
self.assertEqual(session.status, GameSession.Status.ACTIVE)
```
**Step 2: Run test to verify it fails**
```bash
.venv/bin/python manage.py test lobby.tests.GameSessionGameTypeTest --verbosity=2
```
Expected: `AttributeError: type object 'GameSession' has no attribute 'game_type'`
**Step 3: Modify `fupogfakta/models.py` — add `game_type` and `ACTIVE` status**
In the `GameSession` class, add `game_type` field and `ACTIVE` to the Status enum:
```python
class Status(models.TextChoices):
LOBBY = "lobby", "Lobby"
ACTIVE = "active", "Aktiv" # ADD THIS — game is running (state detail in GameRun)
LIE = "lie", "Løgnfase" # keep for now — removed in Batch 6
GUESS = "guess", "Gættefase" # keep for now
REVEAL = "reveal", "Reveal" # keep for now
FINISHED = "finished", "Afsluttet"
```
And add the field (after `current_round`):
```python
game_type = models.CharField(max_length=50, default='fupogfakta')
```
**Step 4: Generate and apply migration**
```bash
.venv/bin/python manage.py makemigrations fupogfakta --name gamesession_game_type_active_status
.venv/bin/python manage.py migrate
```
**Step 5: Run test to verify it passes**
```bash
.venv/bin/python manage.py test lobby.tests.GameSessionGameTypeTest --verbosity=2
```
Expected: PASS.
**Step 6: Run full test suite — must stay green**
```bash
.venv/bin/python manage.py test lobby realtime --verbosity=1
```
Expected: all passing.
**Step 7: Commit**
```bash
git add fupogfakta/models.py fupogfakta/migrations/ lobby/tests.py
git commit -m "feat(platform): add game_type field and ACTIVE status to GameSession"
```
---
### Task 5: Create `GameDriver` interface
**Files:**
- Create: `lobby/driver.py`
**Step 1: Write failing test**
```python
# lobby/tests.py — add to bottom
class GameDriverInterfaceTest(TestCase):
def test_cannot_instantiate_abstract_driver(self):
from lobby.driver import GameDriver
with self.assertRaises(TypeError):
GameDriver()
def test_phase_result_is_namedtuple(self):
from lobby.driver import PhaseResult
result = PhaseResult(next_state='SOME_STATE', duration_seconds=30, broadcast_payload={'type': 'x'})
self.assertEqual(result.next_state, 'SOME_STATE')
self.assertEqual(result.duration_seconds, 30)
```
**Step 2: Run test to verify it fails**
```bash
.venv/bin/python manage.py test lobby.tests.GameDriverInterfaceTest --verbosity=2
```
Expected: `ModuleNotFoundError: No module named 'lobby.driver'`
**Step 3: Create `lobby/driver.py`**
```python
from abc import ABC, abstractmethod
from typing import NamedTuple
class PhaseResult(NamedTuple):
"""Return value from GameDriver phase methods."""
next_state: str
duration_seconds: int | None # None = no timer, wait for manual trigger
broadcast_payload: dict
class GameDriver(ABC):
"""
Abstract base class for game cartridges.
Each game app implements this and registers via GAME_DRIVERS in settings.
The platform (lobby/) calls these methods; game logic never touches
GameRun directly.
"""
game_type: str # must be set on concrete subclass, e.g. "fupogfakta"
@abstractmethod
def on_game_start(self, session, run, config: dict) -> PhaseResult:
"""Called when host presses Play from LOBBY. Returns initial phase."""
@abstractmethod
def on_timer_expired(self, session, run, config: dict) -> PhaseResult:
"""Called when Celery timer fires for run.current_state. Returns next phase."""
@abstractmethod
def on_exit(self, session, run) -> None:
"""
Called when host presses Exit. Must delete all ephemeral game data
(RoundQuestion, LieAnswer, Guess, LieReaction, Player, GameRun).
GameSession row is kept.
"""
def on_pause(self, session, run) -> None:
"""Default: no-op. Override if game needs to pause internal state."""
def on_resume(self, session, run) -> None:
"""Default: no-op. Override if game needs to handle resume."""
def get_driver(game_type: str) -> GameDriver:
"""
Look up and return a registered GameDriver instance by game_type.
Raises KeyError if game_type is not registered.
Import and call this from the platform task/views.
"""
from django.conf import settings
registry: dict[str, str] = getattr(settings, 'GAME_DRIVERS', {})
if game_type not in registry:
raise KeyError(f"No GameDriver registered for game_type={game_type!r}. "
f"Add it to GAME_DRIVERS in settings.py.")
# Lazy import the driver class from dotted path
module_path, class_name = registry[game_type].rsplit('.', 1)
import importlib
module = importlib.import_module(module_path)
cls = getattr(module, class_name)
return cls()
```
**Step 4: Register driver path in `partyhub/settings.py`**
Add after the CELERY block:
```python
GAME_DRIVERS = {
'fupogfakta': 'fupogfakta.driver.FupOgFaktaDriver',
}
```
**Step 5: Run test to verify it passes**
```bash
.venv/bin/python manage.py test lobby.tests.GameDriverInterfaceTest --verbosity=2
```
Expected: PASS.
**Step 6: Commit**
```bash
git add lobby/driver.py partyhub/settings.py lobby/tests.py
git commit -m "feat(platform): add GameDriver abstract interface and get_driver registry"
```
---
### Task 6: Create `GameRun` model
**Files:**
- Create: `lobby/models.py` (currently empty scaffold)
- Create migration
**Step 1: Write failing test**
```python
# lobby/tests.py
class GameRunModelTest(TestCase):
def setUp(self):
from django.contrib.auth import get_user_model
User = get_user_model()
self.user = User.objects.create_user(username='runhost', password='pw')
self.session = GameSession.objects.create(host=self.user, code='RUNTEST')
def test_create_game_run(self):
from lobby.models import GameRun
run = GameRun.objects.create(
session=self.session,
current_state='LIE_PHASE',
)
self.assertEqual(run.current_state, 'LIE_PHASE')
self.assertFalse(run.is_paused)
self.assertIsNone(run.phase_deadline)
self.assertIsNone(run.celery_task_id)
self.assertEqual(run.state_data, {})
def test_game_run_is_one_to_one_with_session(self):
from lobby.models import GameRun
from django.db import IntegrityError
GameRun.objects.create(session=self.session, current_state='LIE_PHASE')
with self.assertRaises(IntegrityError):
GameRun.objects.create(session=self.session, current_state='GUESS_PHASE')
```
**Step 2: Run test to verify it fails**
```bash
.venv/bin/python manage.py test lobby.tests.GameRunModelTest --verbosity=2
```
Expected: fails — GameRun does not exist.
**Step 3: Read and update `lobby/models.py`**
```python
from django.db import models
from django.contrib.auth import get_user_model
User = get_user_model()
class GameRun(models.Model):
"""
Ephemeral runtime state for an active game session.
Created when host presses Play. Deleted when game ends (exit or finish).
"""
session = models.OneToOneField(
'fupogfakta.GameSession',
on_delete=models.CASCADE,
related_name='run',
)
current_state = models.CharField(max_length=64)
phase_deadline = models.DateTimeField(null=True, blank=True)
is_paused = models.BooleanField(default=False)
paused_remaining_seconds = models.FloatField(null=True, blank=True)
celery_task_id = models.CharField(max_length=255, null=True, blank=True)
# Game-specific snapshot data (config, current question id, etc.)
state_data = models.JSONField(default=dict, blank=True)
created_at = models.DateTimeField(auto_now_add=True)
def __str__(self):
return f"GameRun({self.session.code}, state={self.current_state})"
```
**Step 4: Generate and apply migration**
```bash
.venv/bin/python manage.py makemigrations lobby --name add_gamerun
.venv/bin/python manage.py migrate
```
**Step 5: Run test to verify it passes**
```bash
.venv/bin/python manage.py test lobby.tests.GameRunModelTest --verbosity=2
```
Expected: PASS.
**Step 6: Run full suite**
```bash
.venv/bin/python manage.py test lobby realtime --verbosity=1
```
Expected: all green.
**Step 7: Commit**
```bash
git add lobby/models.py lobby/migrations/ lobby/tests.py
git commit -m "feat(platform): add GameRun model"
```
---
## Batch 3 — Config System
---
### Task 7: BaseGameConfig abstract model + FupOgFaktaConfig
**Files:**
- Create: `lobby/base_config.py`
- Modify: `fupogfakta/models.py`
- Create migration
**Step 1: Write failing test**
```python
# fupogfakta/tests.py (create if empty)
from django.test import TestCase
from django.contrib.auth import get_user_model
User = get_user_model()
class FupOgFaktaConfigTest(TestCase):
def setUp(self):
self.user = User.objects.create_user(username='cfghost', password='pw')
def test_system_default_config_exists_after_migration(self):
from fupogfakta.models import FupOgFaktaConfig
default = FupOgFaktaConfig.objects.filter(user=None, is_default=True).first()
# System default is created by a data migration (Task 8)
self.assertIsNotNone(default)
self.assertEqual(default.num_rounds, 3)
self.assertEqual(default.lie_seconds, 45)
def test_user_can_have_multiple_presets(self):
from fupogfakta.models import FupOgFaktaConfig
FupOgFaktaConfig.objects.create(user=self.user, name='Quick', is_default=True)
FupOgFaktaConfig.objects.create(user=self.user, name='Long', is_default=False)
self.assertEqual(FupOgFaktaConfig.objects.filter(user=self.user).count(), 2)
def test_resolve_config_returns_user_default_if_exists(self):
from fupogfakta.config import resolve_config
user_cfg = FupOgFaktaConfig.objects.create(
user=self.user, name='Mine', is_default=True, lie_seconds=20
)
result = resolve_config(self.user)
self.assertEqual(result['lie_seconds'], 20)
def test_resolve_config_falls_back_to_system_default(self):
from fupogfakta.config import resolve_config
# No user preset created
result = resolve_config(self.user)
self.assertEqual(result['lie_seconds'], 45) # system default
```
**Step 2: Run test to verify it fails**
```bash
.venv/bin/python manage.py test fupogfakta --verbosity=2
```
Expected: module-level failures.
**Step 3: Create `lobby/base_config.py`**
```python
from django.db import models
from django.contrib.auth import get_user_model
User = get_user_model()
class BaseGameConfig(models.Model):
"""
Abstract base for per-user game configuration presets.
Each game cartridge extends this with its own concrete fields.
"""
name = models.CharField(max_length=100, default='Default')
user = models.ForeignKey(
User,
on_delete=models.CASCADE,
null=True,
blank=True,
help_text="null = system default (set in Django admin)",
)
is_default = models.BooleanField(
default=False,
help_text="If True and user=null, this is the system default for this game.",
)
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
abstract = True
```
**Step 4: Add `FupOgFaktaConfig` to `fupogfakta/models.py`**
At the bottom of `fupogfakta/models.py`, add:
```python
from lobby.base_config import BaseGameConfig
class FupOgFaktaConfig(BaseGameConfig):
num_rounds = models.PositiveIntegerField(default=3)
questions_per_round = models.PositiveIntegerField(default=3)
min_players = models.PositiveIntegerField(default=2)
max_players = models.PositiveIntegerField(default=8)
lie_seconds = models.PositiveIntegerField(default=45)
guess_seconds = models.PositiveIntegerField(default=30)
reveal_seconds_per_lie = models.PositiveIntegerField(default=8)
scoreboard_recap_seconds = models.PositiveIntegerField(default=10)
awards_display_seconds = models.PositiveIntegerField(default=15)
# Escalating scoring: index 0 = round 1, 1 = round 2, etc.
points_correct = models.JSONField(default=list)
points_bluff = models.JSONField(default=list)
def points_correct_for_round(self, round_index: int) -> int:
defaults = [1500, 3000, 4500]
pts = self.points_correct or defaults
return pts[min(round_index, len(pts) - 1)]
def points_bluff_for_round(self, round_index: int) -> int:
defaults = [500, 1000, 1500]
pts = self.points_bluff or defaults
return pts[min(round_index, len(pts) - 1)]
class Meta:
verbose_name = "Fup og Fakta config"
ordering = ["name"]
def __str__(self):
owner = self.user.username if self.user else "SYSTEM"
return f"{self.name} ({owner})"
```
**Step 5: Create `fupogfakta/config.py`**
```python
from django.contrib.auth import get_user_model
User = get_user_model()
_CONFIG_FIELDS = [
'num_rounds', 'questions_per_round', 'min_players', 'max_players',
'lie_seconds', 'guess_seconds', 'reveal_seconds_per_lie',
'scoreboard_recap_seconds', 'awards_display_seconds',
'points_correct', 'points_bluff',
]
def resolve_config(user) -> dict:
"""
Return config dict for a user starting a fupogfakta session.
Resolution order:
1. User's is_default=True preset (if any)
2. System default (user=None, is_default=True)
3. Model field defaults (hardcoded)
"""
from fupogfakta.models import FupOgFaktaConfig
cfg = (
FupOgFaktaConfig.objects.filter(user=user, is_default=True).first()
or FupOgFaktaConfig.objects.filter(user=None, is_default=True).first()
)
if cfg is None:
# Fall back to model defaults
return _model_defaults()
return {field: getattr(cfg, field) for field in _CONFIG_FIELDS}
def _model_defaults() -> dict:
from fupogfakta.models import FupOgFaktaConfig
tmp = FupOgFaktaConfig()
return {field: getattr(tmp, field) for field in _CONFIG_FIELDS}
```
**Step 6: Generate migration and data migration for system default**
```bash
.venv/bin/python manage.py makemigrations fupogfakta --name add_fupogfakta_config
```
Then create the data migration manually:
```bash
.venv/bin/python manage.py makemigrations fupogfakta --empty --name seed_system_default_config
```
Edit the generated file to add:
```python
def seed_default(apps, schema_editor):
FupOgFaktaConfig = apps.get_model('fupogfakta', 'FupOgFaktaConfig')
FupOgFaktaConfig.objects.create(
name='System Default',
user=None,
is_default=True,
num_rounds=3,
questions_per_round=3,
min_players=2,
max_players=8,
lie_seconds=45,
guess_seconds=30,
reveal_seconds_per_lie=8,
scoreboard_recap_seconds=10,
awards_display_seconds=15,
points_correct=[1500, 3000, 4500],
points_bluff=[500, 1000, 1500],
)
class Migration(migrations.Migration):
dependencies = [('fupogfakta', '0005_...')] # previous migration name
operations = [migrations.RunPython(seed_default, migrations.RunPython.noop)]
```
```bash
.venv/bin/python manage.py migrate
```
**Step 7: Register in admin — modify `fupogfakta/admin.py`**
```python
from django.contrib import admin
from .models import Category, Question, FupOgFaktaConfig
admin.site.register(Category)
admin.site.register(Question)
@admin.register(FupOgFaktaConfig)
class FupOgFaktaConfigAdmin(admin.ModelAdmin):
list_display = ['name', 'user', 'is_default', 'num_rounds', 'lie_seconds', 'guess_seconds']
list_filter = ['is_default', 'user']
```
**Step 8: Run tests**
```bash
.venv/bin/python manage.py test fupogfakta lobby realtime --verbosity=1
```
Expected: all green.
**Step 9: Commit**
```bash
git add lobby/base_config.py fupogfakta/models.py fupogfakta/migrations/ \
fupogfakta/config.py fupogfakta/admin.py fupogfakta/tests.py
git commit -m "feat(config): add FupOgFaktaConfig model, resolve_config, and system default seed"
```
---
## Batch 4 — FupOgFakta Game Models
---
### Task 8: Add `LieReaction` model + `reveal_order` to RoundQuestion, remove ScoreEvent
**Files:**
- Modify: `fupogfakta/models.py`
- Create migrations
**Step 1: Write failing tests**
```python
# fupogfakta/tests.py — add
class LieReactionModelTest(TestCase):
def setUp(self):
from django.contrib.auth import get_user_model
User = get_user_model()
host = User.objects.create_user(username='rhost', password='pw')
from fupogfakta.models import GameSession, Player, Category, Question, RoundQuestion, LieAnswer
self.session = GameSession.objects.create(host=host, code='REACT1')
self.player = Player.objects.create(session=self.session, nickname='Alice')
self.other = Player.objects.create(session=self.session, nickname='Bob')
cat = Category.objects.create(name='Test', slug='test')
q = Question.objects.create(category=cat, prompt='Q?', correct_answer='A')
self.rq = RoundQuestion.objects.create(session=self.session, round_number=1, question=q, correct_answer='A')
self.lie = LieAnswer.objects.create(round_question=self.rq, player=self.player, text='Fake')
def test_player_can_react_to_lie(self):
from fupogfakta.models import LieReaction
r = LieReaction.objects.create(lie=self.lie, player=self.other, reaction='laugh')
self.assertEqual(r.reaction, 'laugh')
def test_duplicate_reaction_raises(self):
from fupogfakta.models import LieReaction
from django.db import IntegrityError
LieReaction.objects.create(lie=self.lie, player=self.other, reaction='laugh')
with self.assertRaises(IntegrityError):
LieReaction.objects.create(lie=self.lie, player=self.other, reaction='laugh')
def test_round_question_has_reveal_order_field(self):
from fupogfakta.models import RoundQuestion
self.assertIsNone(self.rq.reveal_order)
```
**Step 2: Run test to verify it fails**
```bash
.venv/bin/python manage.py test fupogfakta.tests.LieReactionModelTest --verbosity=2
```
**Step 3: Update `fupogfakta/models.py`**
Remove the entire `ScoreEvent` class.
Add `reveal_order` to `RoundQuestion`:
```python
reveal_order = models.PositiveIntegerField(null=True, blank=True)
```
Add `LieReaction` at the bottom:
```python
class LieReaction(models.Model):
REACTION_CHOICES = [
('laugh', '😂 Laugh'),
('heart', '❤️ Heart'),
('fire', '🔥 Fire'),
('wow', '😮 Wow'),
]
lie = models.ForeignKey(LieAnswer, on_delete=models.CASCADE, related_name='reactions')
player = models.ForeignKey(Player, on_delete=models.CASCADE, related_name='lie_reactions')
reaction = models.CharField(max_length=20, choices=REACTION_CHOICES)
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
unique_together = [('lie', 'player', 'reaction')]
ordering = ['created_at']
```
**Step 4: Generate migration**
```bash
.venv/bin/python manage.py makemigrations fupogfakta --name add_liereaction_reveal_order_remove_scoreevent
.venv/bin/python manage.py migrate
```
**Step 5: Run tests**
```bash
.venv/bin/python manage.py test fupogfakta lobby realtime --verbosity=1
```
Expected: all green (ScoreEvent removal will break `calculate_scores` in lobby/views.py — fix in next step).
**Step 6: Fix `lobby/views.py` — remove ScoreEvent references**
In `lobby/views.py`, the `calculate_scores` view imports and uses `ScoreEvent`. Replace the score-saving logic with direct `Player.score` updates only (no audit model). Remove `ScoreEvent` from imports.
The score_events list building and `ScoreEvent.objects.bulk_create(score_events)` lines should be replaced with just the `player.save()` calls already present. Remove the `score_events` list and `events_created` from the response.
**Step 7: Run full test suite**
```bash
.venv/bin/python manage.py test fupogfakta lobby realtime --verbosity=1
```
Expected: all green.
**Step 8: Commit**
```bash
git add fupogfakta/models.py fupogfakta/migrations/ lobby/views.py fupogfakta/tests.py
git commit -m "feat(fupogfakta): add LieReaction + reveal_order, remove ScoreEvent"
```
---
## Batch 5 — FupOgFakta Game Driver
This is the core game logic. Take it one state at a time.
---
### Task 9: Driver skeleton + LIE_PHASE entry
**Files:**
- Create: `fupogfakta/driver.py`
- Create: `fupogfakta/phases.py` (phase transition helpers)
**Step 1: Write failing test**
```python
# fupogfakta/tests.py — add
class FupOgFaktaDriverStartTest(TestCase):
def setUp(self):
from django.contrib.auth import get_user_model
User = get_user_model()
self.host = User.objects.create_user(username='drvhost', password='pw')
from fupogfakta.models import GameSession, Category, Question, FupOgFaktaConfig
from lobby.models import GameRun
self.session = GameSession.objects.create(host=self.host, code='DRVST1')
cat = Category.objects.create(name='DrvCat', slug='drvcat')
Question.objects.create(category=cat, prompt='Q1?', correct_answer='A1')
Question.objects.create(category=cat, prompt='Q2?', correct_answer='A2')
Question.objects.create(category=cat, prompt='Q3?', correct_answer='A3')
self.run = GameRun.objects.create(session=self.session, current_state='LOBBY')
self.config = {
'num_rounds': 1, 'questions_per_round': 1,
'lie_seconds': 30, 'guess_seconds': 20,
'reveal_seconds_per_lie': 5, 'scoreboard_recap_seconds': 5,
'awards_display_seconds': 5,
'points_correct': [1500], 'points_bluff': [500],
'min_players': 2, 'max_players': 8,
'category_slug': 'drvcat',
}
def test_on_game_start_returns_lie_phase(self):
from fupogfakta.driver import FupOgFaktaDriver
driver = FupOgFaktaDriver()
result = driver.on_game_start(self.session, self.run, self.config)
self.assertEqual(result.next_state, 'LIE_PHASE')
self.assertEqual(result.duration_seconds, 30)
self.assertIn('prompt', result.broadcast_payload)
def test_on_game_start_creates_round_question(self):
from fupogfakta.driver import FupOgFaktaDriver
from fupogfakta.models import RoundQuestion
driver = FupOgFaktaDriver()
driver.on_game_start(self.session, self.run, self.config)
self.assertEqual(RoundQuestion.objects.filter(session=self.session).count(), 1)
```
**Step 2: Run test to verify it fails**
```bash
.venv/bin/python manage.py test fupogfakta.tests.FupOgFaktaDriverStartTest --verbosity=2
```
**Step 3: Create `fupogfakta/driver.py`**
```python
import random
import logging
from lobby.driver import GameDriver, PhaseResult
logger = logging.getLogger(__name__)
class FupOgFaktaDriver(GameDriver):
game_type = 'fupogfakta'
def on_game_start(self, session, run, config: dict) -> PhaseResult:
"""Select first question and enter LIE_PHASE."""
from fupogfakta.phases import enter_lie_phase
return enter_lie_phase(session, run, config, round_number=1, question_number=1)
def on_timer_expired(self, session, run, config: dict) -> PhaseResult:
"""Dispatch to the correct next-state handler based on current_state."""
from fupogfakta import phases
state = run.current_state
state_data = run.state_data
if state == 'LIE_PHASE':
return phases.lie_phase_expired(session, run, config)
if state == 'GUESS_PHASE':
return phases.guess_phase_expired(session, run, config)
if state.startswith('REVEAL_LIE_'):
return phases.reveal_lie_expired(session, run, config)
if state == 'REVEAL_TRUTH':
return phases.reveal_truth_expired(session, run, config)
if state == 'SCOREBOARD_RECAP':
return phases.scoreboard_recap_expired(session, run, config)
if state == 'POST_GAME_AWARDS':
return phases.post_game_awards_expired(session, run, config)
logger.error("FupOgFaktaDriver: unknown state %r for run %s", state, run.pk)
return PhaseResult('FINISHED', None, {'type': 'phase.game_over'})
def on_exit(self, session, run) -> None:
"""Delete all ephemeral game data. Keep GameSession row."""
from fupogfakta.models import RoundQuestion, LieAnswer, Guess, LieReaction, Player
from lobby.models import GameRun
# Cascade deletes handle LieAnswer, Guess, LieReaction via RoundQuestion
RoundQuestion.objects.filter(session=session).delete()
Player.objects.filter(session=session).delete()
GameRun.objects.filter(session=session).delete()
session.status = session.Status.FINISHED
session.save(update_fields=['status'])
```
**Step 4: Create `fupogfakta/phases.py`**
Start with just `enter_lie_phase` — other phases added in subsequent tasks:
```python
import random
import logging
from lobby.driver import PhaseResult
logger = logging.getLogger(__name__)
def _pick_question(session, config: dict, round_number: int):
"""Pick a random unused question for this session from the configured category."""
from fupogfakta.models import Category, Question, RoundQuestion
category = Category.objects.get(slug=config['category_slug'])
used_ids = RoundQuestion.objects.filter(session=session).values_list('question_id', flat=True)
available = list(
Question.objects.filter(category=category, is_active=True).exclude(pk__in=used_ids)
)
if not available:
raise ValueError(f"No available questions for category {config['category_slug']!r}")
return random.choice(available)
def enter_lie_phase(session, run, config: dict, round_number: int, question_number: int) -> PhaseResult:
"""Select a question, create RoundQuestion, return LIE_PHASE result."""
from fupogfakta.models import RoundQuestion
question = _pick_question(session, config, round_number)
rq = RoundQuestion.objects.create(
session=session,
round_number=round_number,
question=question,
correct_answer=question.correct_answer,
)
run.state_data = {
**run.state_data,
'round_number': round_number,
'question_number': question_number,
'round_question_id': rq.id,
}
run.save(update_fields=['state_data'])
return PhaseResult(
next_state='LIE_PHASE',
duration_seconds=config['lie_seconds'],
broadcast_payload={
'type': 'phase.lie_started',
'round_question_id': rq.id,
'prompt': question.prompt,
'round_number': round_number,
'question_number': question_number,
'lie_seconds': config['lie_seconds'],
},
)
def lie_phase_expired(session, run, config: dict) -> PhaseResult:
"""Timer ran out in LIE_PHASE. Mix answers and enter GUESS_PHASE."""
from fupogfakta.models import RoundQuestion, LieAnswer
rq_id = run.state_data['round_question_id']
rq = RoundQuestion.objects.get(pk=rq_id)
# Mix answers (dedup, shuffle)
lie_texts = list(LieAnswer.objects.filter(round_question=rq).values_list('text', flat=True))
seen = set()
answers = []
for text in [rq.correct_answer, *lie_texts]:
norm = text.strip().casefold()
if norm and norm not in seen:
seen.add(norm)
answers.append(text.strip())
random.shuffle(answers)
rq.mixed_answers = answers
rq.save(update_fields=['mixed_answers'])
return PhaseResult(
next_state='GUESS_PHASE',
duration_seconds=config['guess_seconds'],
broadcast_payload={
'type': 'phase.guess_started',
'round_question_id': rq.id,
'answers': [{'text': t} for t in answers],
'guess_seconds': config['guess_seconds'],
},
)
def guess_phase_expired(session, run, config: dict) -> PhaseResult:
"""Timer ran out in GUESS_PHASE. Compute reveal order, enter first REVEAL_LIE or REVEAL_TRUTH."""
from fupogfakta.models import RoundQuestion, LieAnswer, Guess
rq_id = run.state_data['round_question_id']
rq = RoundQuestion.objects.prefetch_related('lies', 'guesses').get(pk=rq_id)
# Determine which lies had at least one guesser
guessed_lie_ids = set(
Guess.objects.filter(round_question=rq, is_correct=False)
.exclude(fooled_player=None)
.values_list('fooled_player_id', flat=True)
)
# Map: liar player_id → lie
lie_map = {lie.player_id: lie for lie in rq.lies.all()}
reveal_lies = [lie for pid, lie in lie_map.items() if any(
g.fooled_player_id == pid for g in rq.guesses.all()
)]
# Assign reveal_order to each lie being revealed
for i, lie in enumerate(reveal_lies):
lie.reveal_order = i
lie.save(update_fields=['reveal_order'])
run.state_data = {
**run.state_data,
'reveal_lies': [lie.id for lie in reveal_lies],
'reveal_index': 0,
}
run.save(update_fields=['state_data'])
if reveal_lies:
return _reveal_lie_result(rq, reveal_lies[0], config, run.state_data)
else:
return _reveal_truth_result(rq, config)
def reveal_lie_expired(session, run, config: dict) -> PhaseResult:
"""A lie has been shown. Score the liar. Move to next lie or truth."""
from fupogfakta.models import LieAnswer, Guess, Player, RoundQuestion
rq_id = run.state_data['round_question_id']
reveal_lies = run.state_data['reveal_lies']
reveal_index = run.state_data['reveal_index']
round_number = run.state_data['round_number']
# Score the liar for this lie
lie = LieAnswer.objects.get(pk=reveal_lies[reveal_index])
guessers = list(Guess.objects.filter(round_question_id=rq_id, fooled_player=lie.player).select_related('player'))
points_bluff = _points_bluff(config, round_number)
if guessers:
delta = points_bluff * len(guessers)
lie.player.score += delta
lie.player.save(update_fields=['score'])
# Next reveal
next_index = reveal_index + 1
run.state_data = {**run.state_data, 'reveal_index': next_index}
run.save(update_fields=['state_data'])
rq = RoundQuestion.objects.get(pk=rq_id)
if next_index < len(reveal_lies):
next_lie = LieAnswer.objects.get(pk=reveal_lies[next_index])
return _reveal_lie_result(rq, next_lie, config, run.state_data)
else:
return _reveal_truth_result(rq, config)
def reveal_truth_expired(session, run, config: dict) -> PhaseResult:
"""Truth revealed. Score correct guessers. Enter SCOREBOARD_RECAP."""
from fupogfakta.models import Guess, RoundQuestion
rq_id = run.state_data['round_question_id']
round_number = run.state_data['round_number']
rq = RoundQuestion.objects.get(pk=rq_id)
correct_guessers = list(Guess.objects.filter(round_question=rq, is_correct=True).select_related('player'))
points_correct = _points_correct(config, round_number)
for guess in correct_guessers:
guess.player.score += points_correct
guess.player.save(update_fields=['score'])
leaderboard = _leaderboard(session)
return PhaseResult(
next_state='SCOREBOARD_RECAP',
duration_seconds=config['scoreboard_recap_seconds'],
broadcast_payload={
'type': 'phase.scoreboard',
'leaderboard': leaderboard,
'round_number': round_number,
},
)
def scoreboard_recap_expired(session, run, config: dict) -> PhaseResult:
"""Scoreboard shown. Advance to next question, next round, or post-game awards."""
round_number = run.state_data['round_number']
question_number = run.state_data['question_number']
num_rounds = config['num_rounds']
questions_per_round = config['questions_per_round']
if question_number < questions_per_round:
# Next question in same round
return enter_lie_phase(session, run, config, round_number, question_number + 1)
elif round_number < num_rounds:
# Next round
return enter_lie_phase(session, run, config, round_number + 1, 1)
else:
# All rounds done → post-game awards
return _post_game_awards_result(session, config)
def post_game_awards_expired(session, run, config: dict) -> PhaseResult:
"""Awards shown. Game over."""
return PhaseResult(
next_state='FINISHED',
duration_seconds=None,
broadcast_payload={
'type': 'phase.game_over',
'leaderboard': _leaderboard(session),
},
)
# ── helpers ──────────────────────────────────────────────────────────────────
def _points_correct(config: dict, round_number: int) -> int:
pts = config.get('points_correct') or [1500, 3000, 4500]
return pts[min(round_number - 1, len(pts) - 1)]
def _points_bluff(config: dict, round_number: int) -> int:
pts = config.get('points_bluff') or [500, 1000, 1500]
return pts[min(round_number - 1, len(pts) - 1)]
def _leaderboard(session) -> list:
from fupogfakta.models import Player
return list(
Player.objects.filter(session=session)
.order_by('-score', 'nickname')
.values('id', 'nickname', 'score')
)
def _reveal_lie_result(rq, lie, config: dict, state_data: dict) -> PhaseResult:
from fupogfakta.models import Guess
guessers = list(
Guess.objects.filter(round_question=rq, fooled_player=lie.player)
.select_related('player')
.values('player__nickname')
)
return PhaseResult(
next_state=f'REVEAL_LIE_{state_data["reveal_index"]}',
duration_seconds=config['reveal_seconds_per_lie'],
broadcast_payload={
'type': 'phase.reveal_lie',
'lie_text': lie.text,
'author': lie.player.nickname,
'guessers': [g['player__nickname'] for g in guessers],
'guesser_count': len(guessers),
},
)
def _reveal_truth_result(rq, config: dict) -> PhaseResult:
from fupogfakta.models import Guess
correct_guessers = list(
Guess.objects.filter(round_question=rq, is_correct=True)
.select_related('player')
.values('player__nickname')
)
return PhaseResult(
next_state='REVEAL_TRUTH',
duration_seconds=config['reveal_seconds_per_lie'],
broadcast_payload={
'type': 'phase.reveal_truth',
'correct_answer': rq.correct_answer,
'correct_guessers': [g['player__nickname'] for g in correct_guessers],
},
)
def _post_game_awards_result(session, config: dict) -> PhaseResult:
from fupogfakta.models import LieReaction, LieAnswer
from django.db.models import Count
# Most laughs overall per player
top_laugh = (
LieReaction.objects.filter(lie__round_question__session=session, reaction='laugh')
.values('lie__player__nickname')
.annotate(total=Count('id'))
.order_by('-total')
.first()
)
# Most hearts on a single lie
top_heart = (
LieReaction.objects.filter(lie__round_question__session=session, reaction='heart')
.values('lie__player__nickname', 'lie__text')
.annotate(total=Count('id'))
.order_by('-total')
.first()
)
awards = []
if top_laugh:
awards.append({
'award': 'most_hilarious',
'label': 'Most Hilarious Liar 😂',
'winner': top_laugh['lie__player__nickname'],
'count': top_laugh['total'],
})
if top_heart:
awards.append({
'award': 'most_beloved',
'label': 'Most Beloved Lie ❤️',
'winner': top_heart['lie__player__nickname'],
'lie': top_heart['lie__text'],
'count': top_heart['total'],
})
return PhaseResult(
next_state='POST_GAME_AWARDS',
duration_seconds=config['awards_display_seconds'],
broadcast_payload={
'type': 'phase.awards',
'awards': awards,
'leaderboard': _leaderboard(session),
},
)
```
**Step 5: Run tests**
```bash
.venv/bin/python manage.py test fupogfakta.tests.FupOgFaktaDriverStartTest --verbosity=2
```
Expected: PASS.
**Step 6: Run full suite**
```bash
.venv/bin/python manage.py test fupogfakta lobby realtime --verbosity=1
```
Expected: all green.
**Step 7: Commit**
```bash
git add fupogfakta/driver.py fupogfakta/phases.py fupogfakta/tests.py
git commit -m "feat(fupogfakta): implement game driver and all phase transitions"
```
---
## Batch 6 — Platform REST Endpoints + Celery Wiring
---
### Task 10: Wire `handle_timer_expired` to the driver
**Files:**
- Modify: `lobby/tasks.py`
**Step 1: Write failing test**
```python
# lobby/tests.py — add
class TimerTaskDispatchTest(TestCase):
def setUp(self):
from django.contrib.auth import get_user_model
User = get_user_model()
self.host = User.objects.create_user(username='taskhost', password='pw')
from fupogfakta.models import GameSession, Category, Question, Player
from lobby.models import GameRun
self.session = GameSession.objects.create(host=self.host, code='TASKTS')
self.cat = Category.objects.create(name='TskCat', slug='tskcat')
for i in range(3):
Question.objects.create(category=self.cat, prompt=f'Q{i}', correct_answer=f'A{i}')
Player.objects.create(session=self.session, nickname='P1')
Player.objects.create(session=self.session, nickname='P2')
self.run = GameRun.objects.create(
session=self.session,
current_state='LIE_PHASE',
state_data={
'round_number': 1, 'question_number': 1,
'round_question_id': None, # set below
'config': {
'num_rounds': 1, 'questions_per_round': 1,
'lie_seconds': 30, 'guess_seconds': 20,
'reveal_seconds_per_lie': 5, 'scoreboard_recap_seconds': 5,
'awards_display_seconds': 5,
'points_correct': [1500], 'points_bluff': [500],
'min_players': 2, 'max_players': 8,
'category_slug': 'tskcat',
},
},
)
from fupogfakta.models import RoundQuestion, Question as Q
q = Q.objects.filter(category=self.cat).first()
rq = RoundQuestion.objects.create(session=self.session, round_number=1, question=q, correct_answer=q.correct_answer)
self.run.state_data['round_question_id'] = rq.id
self.run.save(update_fields=['state_data'])
self.rq = rq
def test_task_transitions_lie_to_guess(self):
from lobby.tasks import handle_timer_expired
from lobby.models import GameRun
handle_timer_expired(run_id=self.run.pk, expected_state='LIE_PHASE')
self.run.refresh_from_db()
self.assertEqual(self.run.current_state, 'GUESS_PHASE')
def test_stale_task_does_not_transition(self):
from lobby.tasks import handle_timer_expired
from lobby.models import GameRun
handle_timer_expired(run_id=self.run.pk, expected_state='WRONG_STATE')
self.run.refresh_from_db()
self.assertEqual(self.run.current_state, 'LIE_PHASE') # unchanged
```
**Step 2: Update `lobby/tasks.py` — full implementation**
```python
import logging
from celery import shared_task
from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer
logger = logging.getLogger(__name__)
def _schedule_next(run, duration_seconds: int) -> str:
"""Schedule handle_timer_expired for run after duration_seconds. Returns task id."""
from datetime import timedelta
from django.utils import timezone
eta = timezone.now() + timedelta(seconds=duration_seconds)
result = handle_timer_expired.apply_async(
kwargs={'run_id': run.pk, 'expected_state': run.current_state},
eta=eta,
)
return result.id
def _broadcast(session_code: str, payload: dict) -> None:
channel_layer = get_channel_layer()
group_name = f"game_{session_code.upper()}"
async_to_sync(channel_layer.group_send)(
group_name,
{'type': 'phase_event', 'payload': payload},
)
@shared_task(bind=True, max_retries=0)
def handle_timer_expired(self, run_id: int, expected_state: str) -> None:
from lobby.models import GameRun
from lobby.driver import get_driver
from django.utils import timezone
try:
run = GameRun.objects.select_related('session').get(pk=run_id)
except GameRun.DoesNotExist:
logger.info("handle_timer_expired: run %s not found — stale task", run_id)
return
if run.current_state != expected_state:
logger.info("handle_timer_expired: run %s state mismatch (%r != %r) — stale", run_id, run.current_state, expected_state)
return
if run.is_paused:
logger.info("handle_timer_expired: run %s is paused — ignoring", run_id)
return
config = run.state_data.get('config', {})
driver = get_driver(run.session.game_type)
if run.current_state == 'FINISHED':
# Clean up
driver.on_exit(run.session, run)
return
result = driver.on_timer_expired(run.session, run, config)
if result.next_state == 'FINISHED':
_broadcast(run.session.code, result.broadcast_payload)
driver.on_exit(run.session, run)
return
run.current_state = result.next_state
run.phase_deadline = timezone.now() + __import__('datetime').timedelta(seconds=result.duration_seconds) if result.duration_seconds else None
run.celery_task_id = None
run.save(update_fields=['current_state', 'phase_deadline', 'celery_task_id', 'state_data'])
if result.duration_seconds:
task_id = _schedule_next(run, result.duration_seconds)
run.celery_task_id = task_id
run.save(update_fields=['celery_task_id'])
_broadcast(run.session.code, result.broadcast_payload)
```
**Step 3: Run tests**
```bash
.venv/bin/python manage.py test lobby.tests.TimerTaskDispatchTest --verbosity=2
```
Expected: PASS (CELERY_TASK_ALWAYS_EAGER=True in test mode).
**Step 4: Commit**
```bash
git add lobby/tasks.py lobby/tests.py
git commit -m "feat(platform): wire handle_timer_expired to GameDriver dispatch"
```
---
### Task 11: Platform play/pause/exit endpoints
**Files:**
- Modify: `lobby/views.py` — add play, pause, exit views
- Modify: `lobby/urls.py` — register routes
**Step 1: Write failing tests**
```python
# lobby/tests.py — add
class PlayPauseExitTest(TestCase):
def setUp(self):
from django.contrib.auth import get_user_model
User = get_user_model()
self.host = User.objects.create_user(username='ppe_host', password='pw')
self.client.force_login(self.host)
from fupogfakta.models import GameSession, Category, Question, Player
from fupogfakta.models import FupOgFaktaConfig
self.session = GameSession.objects.create(host=self.host, code='PPETEST')
cat = Category.objects.create(name='PPECat', slug='ppecat')
for i in range(3):
Question.objects.create(category=cat, prompt=f'Q{i}', correct_answer=f'A{i}')
Player.objects.create(session=self.session, nickname='P1')
Player.objects.create(session=self.session, nickname='P2')
def test_play_creates_game_run_and_transitions_to_active(self):
from lobby.models import GameRun
resp = self.client.post(f'/lobby/sessions/PPETEST/play',
content_type='application/json',
data={'category_slug': 'ppecat'})
self.assertEqual(resp.status_code, 201)
self.assertTrue(GameRun.objects.filter(session__code='PPETEST').exists())
def test_pause_sets_is_paused(self):
from lobby.models import GameRun
self.client.post(f'/lobby/sessions/PPETEST/play',
content_type='application/json',
data={'category_slug': 'ppecat'})
resp = self.client.post('/lobby/sessions/PPETEST/pause')
self.assertEqual(resp.status_code, 200)
run = GameRun.objects.get(session__code='PPETEST')
self.assertTrue(run.is_paused)
def test_exit_deletes_game_run(self):
from lobby.models import GameRun
self.client.post('/lobby/sessions/PPETEST/play',
content_type='application/json',
data={'category_slug': 'ppecat'})
resp = self.client.post('/lobby/sessions/PPETEST/exit')
self.assertEqual(resp.status_code, 200)
self.assertFalse(GameRun.objects.filter(session__code='PPETEST').exists())
```
**Step 2: Add views to `lobby/views.py`**
Add these three views at the bottom:
```python
@require_POST
@login_required
def play_session(request: HttpRequest, code: str) -> JsonResponse:
"""Start game (or resume if paused)."""
from lobby.models import GameRun
from lobby.driver import get_driver
from lobby.tasks import _schedule_next, _broadcast
from fupogfakta.config import resolve_config
from django.utils import timezone
session_code = _normalize_session_code(code)
try:
session = GameSession.objects.get(code=session_code)
except GameSession.DoesNotExist:
return JsonResponse({'error': 'Session not found'}, status=404)
if session.host_id != request.user.id:
return JsonResponse({'error': 'Only host can start game'}, status=403)
# Resume from pause
run = GameRun.objects.filter(session=session).first()
if run and run.is_paused:
remaining = run.paused_remaining_seconds or 0
run.phase_deadline = timezone.now() + __import__('datetime').timedelta(seconds=remaining)
run.is_paused = False
run.paused_remaining_seconds = None
run.save(update_fields=['phase_deadline', 'is_paused', 'paused_remaining_seconds'])
task_id = _schedule_next(run, int(remaining))
run.celery_task_id = task_id
run.save(update_fields=['celery_task_id'])
_broadcast(session.code, {'type': 'phase.resumed', 'phase_deadline': run.phase_deadline.isoformat()})
return JsonResponse({'status': 'resumed'})
if run:
return JsonResponse({'error': 'Game already running'}, status=409)
# Fresh start
payload = _json_body(request)
category_slug = str(payload.get('category_slug', '')).strip()
if not category_slug:
return JsonResponse({'error': 'category_slug required'}, status=400)
config = resolve_config(request.user)
config['category_slug'] = category_slug
run = GameRun.objects.create(
session=session,
current_state='LOBBY',
state_data={'config': config},
)
driver = get_driver(session.game_type)
result = driver.on_game_start(session, run, config)
run.current_state = result.next_state
run.state_data['config'] = config
run.save(update_fields=['current_state', 'state_data'])
if result.duration_seconds:
run.phase_deadline = timezone.now() + __import__('datetime').timedelta(seconds=result.duration_seconds)
task_id = _schedule_next(run, result.duration_seconds)
run.celery_task_id = task_id
run.save(update_fields=['phase_deadline', 'celery_task_id'])
session.status = GameSession.Status.ACTIVE
session.save(update_fields=['status'])
_broadcast(session.code, result.broadcast_payload)
return JsonResponse({'status': 'started', 'state': run.current_state}, status=201)
@require_POST
@login_required
def pause_session(request: HttpRequest, code: str) -> JsonResponse:
from lobby.models import GameRun
from lobby.tasks import _broadcast
from django.utils import timezone
from celery.app.control import Control
session_code = _normalize_session_code(code)
try:
session = GameSession.objects.get(code=session_code)
except GameSession.DoesNotExist:
return JsonResponse({'error': 'Session not found'}, status=404)
if session.host_id != request.user.id:
return JsonResponse({'error': 'Only host can pause'}, status=403)
try:
run = GameRun.objects.get(session=session)
except GameRun.DoesNotExist:
return JsonResponse({'error': 'Game not running'}, status=400)
if run.is_paused:
return JsonResponse({'error': 'Already paused'}, status=400)
# Revoke Celery task
if run.celery_task_id:
from partyhub.celery import app as celery_app
celery_app.control.revoke(run.celery_task_id, terminate=True)
remaining = (run.phase_deadline - timezone.now()).total_seconds() if run.phase_deadline else 0
run.is_paused = True
run.paused_remaining_seconds = max(remaining, 0)
run.celery_task_id = None
run.save(update_fields=['is_paused', 'paused_remaining_seconds', 'celery_task_id'])
_broadcast(session.code, {'type': 'phase.paused', 'remaining_seconds': run.paused_remaining_seconds})
return JsonResponse({'status': 'paused', 'remaining_seconds': run.paused_remaining_seconds})
@require_POST
@login_required
def exit_session(request: HttpRequest, code: str) -> JsonResponse:
from lobby.models import GameRun
from lobby.driver import get_driver
from celery.app.control import Control
session_code = _normalize_session_code(code)
try:
session = GameSession.objects.get(code=session_code)
except GameSession.DoesNotExist:
return JsonResponse({'error': 'Session not found'}, status=404)
if session.host_id != request.user.id:
return JsonResponse({'error': 'Only host can exit'}, status=403)
run = GameRun.objects.filter(session=session).first()
if run:
if run.celery_task_id:
from partyhub.celery import app as celery_app
celery_app.control.revoke(run.celery_task_id, terminate=True)
driver = get_driver(session.game_type)
driver.on_exit(session, run)
return JsonResponse({'status': 'exited'})
```
**Step 3: Add routes to `lobby/urls.py`**
```python
path("sessions/<str:code>/play", views.play_session, name="play_session"),
path("sessions/<str:code>/pause", views.pause_session, name="pause_session"),
path("sessions/<str:code>/exit", views.exit_session, name="exit_session"),
```
**Step 4: Run tests**
```bash
.venv/bin/python manage.py test lobby.tests.PlayPauseExitTest --verbosity=2
```
Expected: PASS.
**Step 5: Run full suite**
```bash
.venv/bin/python manage.py test fupogfakta lobby realtime --verbosity=1
```
**Step 6: Commit**
```bash
git add lobby/views.py lobby/urls.py lobby/tests.py
git commit -m "feat(platform): add play/pause/exit session endpoints"
```
---
### Task 12: FupOgFakta game-specific endpoints (lie, guess, react)
**Files:**
- Create: `fupogfakta/views.py`
- Create: `fupogfakta/urls.py`
- Modify: `partyhub/urls.py`
**Step 1: Write failing tests**
```python
# fupogfakta/tests.py — add
class SubmitLieViewTest(TestCase):
def setUp(self):
# Create session in LIE_PHASE with a RoundQuestion active
from django.contrib.auth import get_user_model
User = get_user_model()
self.host = User.objects.create_user(username='liehost', password='pw')
from fupogfakta.models import GameSession, Player, Category, Question, RoundQuestion
from lobby.models import GameRun
self.session = GameSession.objects.create(host=self.host, code='LIETEST', status='active')
cat = Category.objects.create(name='LieCat', slug='liecat')
q = Question.objects.create(category=cat, prompt='What is?', correct_answer='Real Answer')
self.rq = RoundQuestion.objects.create(session=self.session, round_number=1, question=q, correct_answer='Real Answer')
self.player = Player.objects.create(session=self.session, nickname='Liar')
self.run = GameRun.objects.create(
session=self.session, current_state='LIE_PHASE',
state_data={'round_question_id': self.rq.id, 'config': {'lie_seconds': 45}}
)
def test_submit_lie_succeeds(self):
resp = self.client.post(
'/fupogfakta/LIETEST/lie',
content_type='application/json',
data={'session_token': self.player.session_token, 'text': 'My fake answer'},
)
self.assertEqual(resp.status_code, 201)
def test_submit_correct_answer_as_lie_rejected(self):
resp = self.client.post(
'/fupogfakta/LIETEST/lie',
content_type='application/json',
data={'session_token': self.player.session_token, 'text': 'real answer'},
)
self.assertEqual(resp.status_code, 422)
self.assertIn('lie_matches_correct_answer', resp.json().get('error_code', ''))
```
**Step 2: Create `fupogfakta/views.py`**
```python
import json
from django.http import HttpRequest, JsonResponse
from django.views.decorators.http import require_POST
from fupogfakta.models import GameSession, Player, RoundQuestion, LieAnswer, Guess, LieReaction
from lobby.models import GameRun
def _json_body(request: HttpRequest) -> dict:
try:
return json.loads(request.body) if request.body else {}
except json.JSONDecodeError:
return {}
def _get_active_run_and_player(code: str, session_token: str):
"""Returns (session, run, player) or raises."""
try:
session = GameSession.objects.get(code=code.upper())
except GameSession.DoesNotExist:
return None, None, None
try:
run = GameRun.objects.get(session=session)
except GameRun.DoesNotExist:
return session, None, None
try:
player = Player.objects.get(session=session, session_token=session_token)
except Player.DoesNotExist:
return session, run, None
return session, run, player
@require_POST
def submit_lie(request: HttpRequest, code: str) -> JsonResponse:
payload = _json_body(request)
session_token = str(payload.get('session_token', '')).strip()
lie_text = str(payload.get('text', '')).strip()
if not session_token:
return JsonResponse({'error': 'session_token required'}, status=400)
if not lie_text or len(lie_text) > 255:
return JsonResponse({'error': 'text must be 1-255 characters'}, status=400)
session, run, player = _get_active_run_and_player(code, session_token)
if not session:
return JsonResponse({'error': 'Session not found'}, status=404)
if not run:
return JsonResponse({'error': 'Game not running'}, status=400)
if not player:
return JsonResponse({'error': 'Invalid session_token'}, status=403)
if run.current_state != 'LIE_PHASE':
return JsonResponse({'error': 'Not in lie phase'}, status=400)
rq_id = run.state_data.get('round_question_id')
try:
rq = RoundQuestion.objects.get(pk=rq_id, session=session)
except RoundQuestion.DoesNotExist:
return JsonResponse({'error': 'No active question'}, status=400)
# Reject if lie matches correct answer (case-insensitive)
if lie_text.strip().casefold() == rq.correct_answer.strip().casefold():
return JsonResponse({
'error': 'Your lie matches the correct answer — try something else!',
'error_code': 'lie_matches_correct_answer',
}, status=422)
from django.db import IntegrityError
try:
lie = LieAnswer.objects.create(round_question=rq, player=player, text=lie_text)
except IntegrityError:
return JsonResponse({'error': 'Lie already submitted'}, status=409)
# Broadcast progress (anonymous count only)
from realtime.broadcast import sync_broadcast_phase_event
total_players = Player.objects.filter(session=session).count()
submitted = LieAnswer.objects.filter(round_question=rq).count()
sync_broadcast_phase_event(session.code, 'phase.lie_progress', {
'submitted': submitted,
'total': total_players,
})
return JsonResponse({'lie_id': lie.id}, status=201)
@require_POST
def submit_guess(request: HttpRequest, code: str) -> JsonResponse:
payload = _json_body(request)
session_token = str(payload.get('session_token', '')).strip()
selected_text = str(payload.get('selected_text', '')).strip()
if not session_token:
return JsonResponse({'error': 'session_token required'}, status=400)
if not selected_text:
return JsonResponse({'error': 'selected_text required'}, status=400)
session, run, player = _get_active_run_and_player(code, session_token)
if not session:
return JsonResponse({'error': 'Session not found'}, status=404)
if not run:
return JsonResponse({'error': 'Game not running'}, status=400)
if not player:
return JsonResponse({'error': 'Invalid session_token'}, status=403)
if run.current_state != 'GUESS_PHASE':
return JsonResponse({'error': 'Not in guess phase'}, status=400)
rq_id = run.state_data.get('round_question_id')
rq = RoundQuestion.objects.get(pk=rq_id, session=session)
allowed = {t.strip().casefold() for t in rq.mixed_answers}
if selected_text.casefold() not in allowed:
return JsonResponse({'error': 'Answer not in this round'}, status=400)
correct_normalized = rq.correct_answer.strip().casefold()
is_correct = selected_text.casefold() == correct_normalized
fooled_player_id = None
if not is_correct:
fooled_player_id = (
LieAnswer.objects.filter(round_question=rq, text__iexact=selected_text)
.values_list('player_id', flat=True)
.first()
)
from django.db import IntegrityError
try:
guess = Guess.objects.create(
round_question=rq,
player=player,
selected_text=selected_text,
is_correct=is_correct,
fooled_player_id=fooled_player_id,
)
except IntegrityError:
return JsonResponse({'error': 'Guess already submitted'}, status=409)
return JsonResponse({'guess_id': guess.id, 'is_correct': is_correct}, status=201)
@require_POST
def submit_reaction(request: HttpRequest, code: str) -> JsonResponse:
payload = _json_body(request)
session_token = str(payload.get('session_token', '')).strip()
lie_id = payload.get('lie_id')
reaction = str(payload.get('reaction', '')).strip()
VALID_REACTIONS = {'laugh', 'heart', 'fire', 'wow'}
if not session_token:
return JsonResponse({'error': 'session_token required'}, status=400)
if reaction not in VALID_REACTIONS:
return JsonResponse({'error': f'reaction must be one of {sorted(VALID_REACTIONS)}'}, status=400)
session, run, player = _get_active_run_and_player(code, session_token)
if not session:
return JsonResponse({'error': 'Session not found'}, status=404)
if not run or run.current_state != 'GUESS_PHASE':
return JsonResponse({'error': 'Reactions only allowed during guess phase'}, status=400)
if not player:
return JsonResponse({'error': 'Invalid session_token'}, status=403)
# Cannot react to own lie
try:
lie = LieAnswer.objects.get(pk=lie_id, round_question__session=session)
except LieAnswer.DoesNotExist:
return JsonResponse({'error': 'Lie not found'}, status=404)
if lie.player_id == player.id:
return JsonResponse({'error': 'Cannot react to your own lie'}, status=400)
from django.db import IntegrityError
try:
LieReaction.objects.create(lie=lie, player=player, reaction=reaction)
except IntegrityError:
# Already reacted with this emoji — idempotent, return ok
pass
return JsonResponse({'status': 'ok'})
```
**Step 3: Create `fupogfakta/urls.py`**
```python
from django.urls import path
from . import views
app_name = 'fupogfakta'
urlpatterns = [
path('<str:code>/lie', views.submit_lie, name='submit_lie'),
path('<str:code>/guess', views.submit_guess, name='submit_guess'),
path('<str:code>/react', views.submit_reaction, name='submit_reaction'),
]
```
**Step 4: Register in `partyhub/urls.py`**
Read the file first, then add:
```python
path('fupogfakta/', include('fupogfakta.urls')),
```
**Step 5: Run tests**
```bash
.venv/bin/python manage.py test fupogfakta.tests.SubmitLieViewTest --verbosity=2
```
Expected: PASS.
**Step 6: Commit**
```bash
git add fupogfakta/views.py fupogfakta/urls.py partyhub/urls.py fupogfakta/tests.py
git commit -m "feat(fupogfakta): add lie/guess/react endpoints"
```
---
## Batch 7 — Frontend Game Screens
The Angular host-shell and player-shell currently work as developer control panels. Replace them with proper game screens that react to WebSocket events.
---
### Task 13: WebSocket state machine in Angular
**Files:**
- Create: `frontend/angular/src/app/game-state.service.ts`
The service opens a WebSocket to `ws/game/{sessionCode}/` and maintains reactive state from `phase.*` events. Both host and player shells subscribe to this service.
```typescript
// game-state.service.ts
import { Injectable, signal } from '@angular/core';
export type GamePhase =
| 'LOBBY' | 'LIE_PHASE' | 'GUESS_PHASE'
| `REVEAL_LIE_${number}` | 'REVEAL_TRUTH'
| 'SCOREBOARD_RECAP' | 'POST_GAME_AWARDS' | 'FINISHED';
export interface GameState {
phase: GamePhase;
sessionCode: string;
payload: Record<string, unknown>;
}
@Injectable({ providedIn: 'root' })
export class GameStateService {
readonly state = signal<GameState>({ phase: 'LOBBY', sessionCode: '', payload: {} });
private ws: WebSocket | null = null;
connect(sessionCode: string, token: string, role: 'player' | 'host') {
const query = role === 'host' ? 'role=host' : `session_token=${token}`;
this.ws = new WebSocket(`ws://${location.host}/ws/game/${sessionCode}/?${query}`);
this.ws.onmessage = (ev) => {
const msg = JSON.parse(ev.data);
this.state.set({ phase: msg.type.replace('phase.', '').toUpperCase() as GamePhase, sessionCode, payload: msg });
};
}
disconnect() { this.ws?.close(); }
ping() { this.ws?.send(JSON.stringify({ type: 'ping' })); }
}
```
**Step 1: Write Vitest test**
```typescript
// frontend/angular/src/app/game-state.service.spec.ts
import { TestBed } from '@angular/core/testing';
import { GameStateService } from './game-state.service';
describe('GameStateService', () => {
it('initialises in LOBBY phase', () => {
const svc = TestBed.inject(GameStateService);
expect(svc.state().phase).toBe('LOBBY');
});
});
```
**Step 2: Run test**
```bash
cd frontend/angular && npm test -- --run 2>&1 | grep -E 'PASS|FAIL|GameState'
```
**Step 3: Rebuild host-shell for real gameplay**
Replace `host-shell.component.ts` content. The host screen shows:
- **LOBBY**: session code large, player list, "Play" button with category select
- **LIE_PHASE**: question prompt, lie-submission progress bar (N/total), countdown timer, Pause button
- **GUESS_PHASE**: answers displayed (no correct answer highlighted), countdown timer, Pause button
- **REVEAL_LIE_\***: lie text, author revealed, guessers listed, score delta animation
- **REVEAL_TRUTH**: correct answer, who guessed right, score deltas
- **SCOREBOARD_RECAP**: full leaderboard
- **POST_GAME_AWARDS**: award cards
- **FINISHED**: final leaderboard, "New Game" button
Host makes REST calls only for: POST `/lobby/sessions/{code}/play`, POST `pause`, POST `exit`.
**Step 4: Rebuild player-shell for real gameplay**
Player screen shows:
- **LOBBY**: "Waiting for host to start…" with players list
- **LIE_PHASE**: text input for lie, submit button, countdown — replaced by "Submitted ✓" once sent
- **GUESS_PHASE**: answer buttons, once selected show emoji reaction buttons for other lies (cannot react to own lie), countdown
- **REVEAL_\*** / **SCOREBOARD** / **AWARDS**: display-only, shows own score delta highlighted
- **FINISHED**: final leaderboard
Player makes REST calls only for: POST `/fupogfakta/{code}/lie`, POST `/fupogfakta/{code}/guess`, POST `/fupogfakta/{code}/react`.
**Step 5: Run Angular tests**
```bash
cd frontend/angular && npm test -- --run 2>&1 | tail -10
```
**Step 6: Commit**
```bash
git add frontend/angular/src/
git commit -m "feat(frontend): rebuild host and player screens as real game UI"
```
---
## Batch 8 — Cleanup
### Task 14: Remove obsolete lobby/views.py endpoints
The old manual-advance endpoints (`start_round`, `show_question`, `submit_lie`, `mix_answers`, `submit_guess`, `calculate_scores`, `reveal_scoreboard`, `finish_game`, `start_next_round`) are now replaced by the driver + platform endpoints. Remove them from `lobby/views.py` and `lobby/urls.py`, then remove or update the tests that covered them.
Run full suite after:
```bash
.venv/bin/python manage.py test fupogfakta lobby realtime --verbosity=1
```
Commit:
```bash
git commit -m "chore: remove obsolete manual-advance lobby endpoints"
```
---
### Task 15: Update TODO.md
Mark completed items, add new backlog items discovered during implementation:
- [ ] Rate limiting on fupogfakta/lie and /guess endpoints
- [ ] Session-code brute-force protection on /join
- [ ] TTS / read-aloud integration (Fase 4)
```bash
git add TODO.md && git commit -m "docs: update TODO after game engine implementation"
```
---
## Running Order Summary
```
Batch 1 — Celery infra (Tasks 1-3) no breaking changes
Batch 2 — GameRun + Driver (Tasks 4-6) additive
Batch 3 — Config system (Task 7) additive
Batch 4 — Game models (Task 8) remove ScoreEvent
Batch 5 — FupOgFakta driver (Task 9) new game logic
Batch 6 — Platform endpoints (Tasks 10-12) new REST API
Batch 7 — Frontend (Tasks 13) replace UI
Batch 8 — Cleanup (Tasks 14-15) remove old code
```
Each batch is independently mergeable. Run `.venv/bin/python manage.py test fupogfakta lobby realtime` before every commit.
Reference in New Issue View Git Blame Copy Permalink