daniel156161
4403978537
Build and Push Docker Container / build-and-push (push) Successful in 10m12s
- Acknowledge start and end without waiting for persistence or dashboards. - Keep runtime creation and finished-game removal on the response path. - Serialize start, turn, and end bookkeeping per game in Quart tasks. - Capture turn thinking before scheduling deferred persistence. - Defer start metrics, logging, and stale-game pruning. - Run legacy history exports in a worker thread. - Record results and end metrics even when history export fails. - Drain pending bookkeeping before closing storage during shutdown. - Test sub-500 ms responses with slow backends and real SQLite. - Cover event ordering, export failures, and rejected user agents.
54 lines
2.0 KiB
Python
54 lines
2.0 KiB
Python
from quart_common.web.logger import await_log, logging
|
|
|
|
from server.database import GameplayDatabase
|
|
from server.GameBoard import GameBoard
|
|
|
|
class GameplayTrackingService:
|
|
def __init__(self, gameplay_database:GameplayDatabase, logger:logging):
|
|
self.gameplay_database = gameplay_database
|
|
self.logger = logger
|
|
|
|
async def record_gameplay_start(self, game_state:dict, game_board:GameBoard) -> None:
|
|
snake_name, snake_version = game_board.get_snake_name_and_version()
|
|
|
|
if self.gameplay_database is None:
|
|
return
|
|
try:
|
|
await self.gameplay_database.record_game_start(
|
|
game_state,
|
|
snake_type=snake_name,
|
|
snake_version=snake_version,
|
|
)
|
|
except Exception as error:
|
|
await await_log(self.logger.warning(f"Gameplay DB start record failed:{error}"))
|
|
|
|
async def record_gameplay_turn(self, game_state:dict, my_move:str, game_board:GameBoard) -> None:
|
|
thinking = self._extract_latest_snake_thinking(game_board)
|
|
await self.record_gameplay_turn_snapshot(game_state, my_move, thinking)
|
|
|
|
async def record_gameplay_turn_snapshot(self, game_state:dict, my_move:str|None, thinking:dict|None) -> None:
|
|
if self.gameplay_database is None:
|
|
return
|
|
try:
|
|
await self.gameplay_database.record_turn(game_state, my_move, thinking)
|
|
except Exception as error:
|
|
await await_log(self.logger.warning(f"Gameplay DB turn record failed:{error}"))
|
|
|
|
async def record_gameplay_end(self, game_state:dict) -> None:
|
|
if self.gameplay_database is None:
|
|
return
|
|
try:
|
|
await self.gameplay_database.record_game_end(game_state)
|
|
except Exception as error:
|
|
await await_log(self.logger.warning(f"Gameplay DB end record failed:{error}"))
|
|
|
|
def _extract_latest_snake_thinking(self, game_board:GameBoard) -> dict|None:
|
|
try:
|
|
history = game_board.snake_class.get_history()
|
|
except Exception:
|
|
return None
|
|
if not isinstance(history, list) or len(history) == 0:
|
|
return None
|
|
latest = history[-1]
|
|
return latest if isinstance(latest, dict) else None
|