# retoor <retoor@molodetz.nl>
from typing import Annotated
from fastapi import APIRouter, Form, Request
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
from devplacepy.models import (
GameCosmeticForm,
GameInfraForm,
GameLegacyForm,
GameMasteryForm,
GamePerkForm,
GamePlantForm,
GameQuestForm,
GameSlotForm,
)
from devplacepy.database import mark_notifications_read_by_target
from devplacepy.responses import json_error, respond, wants_json
from devplacepy.schemas import GameLeaderboardOut, GameStateOut
from devplacepy.services.game import GameError, economy, store
from devplacepy.services.audit import record as audit
from devplacepy.utils import award_rewards, get_current_user, require_user, track_action
from ._shared import action_error, game_seo, notify_farm, state_payload
router = APIRouter()
@router.get("", response_class=HTMLResponse)
async def game_home(request: Request):
user = require_user(request)
mark_notifications_read_by_target(user["uid"], "/game")
farm = state_payload(user)
error = request.query_params.get("error", "")
seo_ctx = game_seo(
request,
"Code Farm",
"Grow software projects on your Code Farm, harvest coins and XP, "
"upgrade your CI, and water your friends' builds.",
)
return respond(
request,
"game.html",
{**seo_ctx, "request": request, "user": user, "farm": farm, "game_error": error},
model=GameStateOut,
)
@router.get("/state")
async def game_state(request: Request):
user = require_user(request)
return JSONResponse(GameStateOut(farm=state_payload(user)).model_dump(mode="json"))
@router.get("/leaderboard")
async def game_leaderboard(request: Request, board: str = "score"):
get_current_user(request)
entries = store.leaderboard_for(board, 25)
return JSONResponse(
GameLeaderboardOut(entries=entries).model_dump(mode="json")
)
async def _respond_action(request: Request, user: dict, fn, on_success=None):
try:
result = fn()
except GameError as exc:
return action_error(request, str(exc), "/game")
if on_success:
on_success(result)
await notify_farm(user.get("username", ""))
if wants_json(request):
return JSONResponse(
GameStateOut(farm=state_payload(user)).model_dump(mode="json")
)
return RedirectResponse(url="/game", status_code=302)
@router.post("/plant")
async def game_plant(request: Request, data: Annotated[GamePlantForm, Form()]):
user = require_user(request)
return await _respond_action(
request, user, lambda: store.plant(user, data.slot, data.crop)
)
@router.post("/harvest")
async def game_harvest(request: Request, data: Annotated[GameSlotForm, Form()]):
user = require_user(request)
def reward(result):
track_action(user["uid"], "harvest")
award_rewards(user["uid"], economy.site_xp_for(result.get("xp", 0)))
return await _respond_action(
request, user, lambda: store.harvest(user, data.slot), reward
)
@router.post("/buy-plot")
async def game_buy_plot(request: Request):
user = require_user(request)
return await _respond_action(request, user, lambda: store.buy_plot(user))
@router.post("/upgrade")
async def game_upgrade(request: Request):
user = require_user(request)
return await _respond_action(request, user, lambda: store.upgrade_ci(user))
@router.post("/fertilize")
async def game_fertilize(request: Request, data: Annotated[GameSlotForm, Form()]):
user = require_user(request)
return await _respond_action(request, user, lambda: store.fertilize(user, data.slot))
@router.post("/daily")
async def game_daily(request: Request):
user = require_user(request)
return await _respond_action(request, user, lambda: store.claim_daily(user))
@router.post("/grant")
async def game_claim_grant(request: Request):
user = require_user(request)
def recorded(result):
audit.record(
request,
"game.grant.claim",
user=user,
metadata=result,
summary=f"{user['username']} claimed a {result['amount']} coin community grant",
)
return await _respond_action(request, user, lambda: store.claim_grant(user), recorded)
@router.post("/perk")
async def game_perk(request: Request, data: Annotated[GamePerkForm, Form()]):
user = require_user(request)
return await _respond_action(
request, user, lambda: store.upgrade_perk(user, data.perk)
)
@router.post("/prestige")
async def game_prestige(request: Request):
user = require_user(request)
def recorded(result):
audit.record(
request,
"game.prestige",
user=user,
metadata=result,
summary=(
f"{user['username']} refactored to prestige {result['prestige']} "
f"for {result['fee']} coins"
),
)
return await _respond_action(request, user, lambda: store.prestige(user), recorded)
@router.post("/legacy")
async def game_legacy(request: Request, data: Annotated[GameLegacyForm, Form()]):
user = require_user(request)
return await _respond_action(
request, user, lambda: store.upgrade_legacy(user, data.key)
)
@router.post("/quests/claim")
async def game_claim_quest(request: Request, data: Annotated[GameQuestForm, Form()]):
user = require_user(request)
def reward(result):
award_rewards(user["uid"], economy.site_xp_for(result.get("reward_xp", 0)))
return await _respond_action(
request, user, lambda: store.claim_quest(user, data.quest, data.scope), reward
)
@router.post("/defense/upgrade")
async def game_upgrade_defense(request: Request):
user = require_user(request)
def reward(result):
track_action(user["uid"], "defense_upgraded")
audit.record(
request,
"game.defense.upgrade",
user=user,
metadata=result,
summary=(
f"{user['username']} bought Code Farm defense level "
f"{result['defense_level']} for {result['spent']} coins"
),
)
return await _respond_action(request, user, lambda: store.upgrade_defense(user), reward)
@router.post("/defense/downgrade")
async def game_downgrade_defense(request: Request):
user = require_user(request)
def recorded(result):
audit.record(
request,
"game.defense.downgrade",
user=user,
metadata=result,
summary=(
f"{user['username']} dropped Code Farm defense to level "
f"{result['defense_level']}"
),
)
return await _respond_action(
request, user, lambda: store.downgrade_defense(user), recorded
)
@router.post("/infrastructure/buy")
async def game_buy_infrastructure(request: Request, data: Annotated[GameInfraForm, Form()]):
user = require_user(request)
def reward(result):
track_action(user["uid"], "infra_bought")
audit.record(
request,
"game.infrastructure.buy",
user=user,
metadata=result,
summary=(
f"{user['username']} bought Code Farm infrastructure {result['key']} "
f"for {result['spent']} coins"
),
)
return await _respond_action(
request, user, lambda: store.buy_infrastructure(user, data.key), reward
)
@router.post("/mastery")
async def game_upgrade_mastery(request: Request, data: Annotated[GameMasteryForm, Form()]):
user = require_user(request)
return await _respond_action(
request, user, lambda: store.upgrade_mastery(user, data.key)
)
@router.post("/cosmetics/buy")
async def game_buy_cosmetic(request: Request, data: Annotated[GameCosmeticForm, Form()]):
user = require_user(request)
def reward(result):
track_action(user["uid"], "cosmetic_bought")
audit.record(
request,
"game.cosmetic.buy",
user=user,
metadata=result,
summary=(
f"{user['username']} bought Code Farm cosmetic {result['key']} "
f"for {result['spent']} coins"
),
)
return await _respond_action(
request, user, lambda: store.buy_cosmetic(user, data.key), reward
)
@router.post("/cosmetics/equip")
async def game_equip_cosmetic(request: Request, data: Annotated[GameCosmeticForm, Form()]):
user = require_user(request)
return await _respond_action(
request, user, lambda: store.equip_title(user, data.key)
)
Reference in New Issue View Git Blame Copy Permalink