"""World content router -- rooms + chunks, for the editor and player client. Implements `docs/architecture/world_persistence_impl.md` §8.1. JWT-authenticated (``CurrentUser``, same as ``app/routers/worlds.py``). All chunk content reads/writes go through ``WorldStore``; this module is HTTP plumbing + authorization only. Authorization (doc §5.3): - **Read** (world visibility, room listing, chunk fetch/index/revisions): same rule as ``GET /api/worlds/{id}`` -- 404 (not 403) if the world is neither PUBLIC nor owned by the caller. See ``_get_world_readable``. - **Write** (room create, chunk PUT/DELETE/rollback): caller must be ``world.owner_id``, matching ``update_world`` -- 403 otherwise, with *no* visibility gate (this deliberately mirrors ``update_world``, which also has no visibility check; see ``_get_world_writable``). """ from __future__ import annotations import base64 import binascii from typing import Annotated from fastapi import APIRouter, Depends, Header, HTTPException, Request, Response from fastapi.responses import JSONResponse from pydantic import BaseModel, Field from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.config import settings from app.database import get_db from app.dependencies import CurrentUser from app.models.user import User from app.models.world import World, WorldVisibility from app.models.world_content import MAIN_ROOM_SLUG, WorldRoom from app.services import chunk_codec from app.services.game_servers import supervisor from app.services.world_store import Bounds, RevisionConflict, RevisionNotAvailable, WorldStore router = APIRouter() DbDep = Annotated[AsyncSession, Depends(get_db)] # world_persistence_impl.md §8.1: "Bounds are capped (default 32x32 chunks # per query) and return 422 beyond that." MAX_BOUNDS_SPAN = 32 # §8.1 batch write cap, beyond WORLD_CHUNK_MAX_BATCH chunks: "Capped at ... # 8 MB." MAX_BATCH_TOTAL_BYTES = 8 * 1024 * 1024 # ─────────────────────────── auth helpers ─────────────────────────── async def _get_world_readable(world_id: int, current_user: User, db: AsyncSession) -> World: result = await db.execute(select(World).where(World.id == world_id)) world = result.scalar_one_or_none() if world is None: raise HTTPException(status_code=404, detail="World not found") if world.visibility != WorldVisibility.PUBLIC and world.owner_id != current_user.id: raise HTTPException(status_code=404, detail="World not found") return world async def _get_world_writable(world_id: int, current_user: User, db: AsyncSession) -> World: result = await db.execute(select(World).where(World.id == world_id)) world = result.scalar_one_or_none() if world is None: raise HTTPException(status_code=404, detail="World not found") if world.owner_id != current_user.id: raise HTTPException(status_code=403, detail="Not owner of this world") return world async def _resolve_room(store: WorldStore, db: AsyncSession, world_id: int, slug: str) -> WorldRoom | None: """world_persistence_impl.md §4.1: lazily create the main room for worlds that predate this feature, on first read/write.""" room = await store.get_room(world_id, slug) if room is None and slug == MAIN_ROOM_SLUG: room = await store.get_or_create_main_room(world_id) await db.commit() return room def _is_live(world_id: int) -> bool: return any(s["world_id"] == str(world_id) for s in supervisor.debug_state()) def _validation_error_body(exc: chunk_codec.ChunkValidationError) -> dict: return {"detail": "chunk validation failed", "reason": exc.reason} def _conflict_body(exc: RevisionConflict) -> dict: body: dict = { "detail": "revision conflict", "current_revision": exc.current_revision, "current_checksum": exc.current_checksum, } if exc.coord is not None: body["coord"] = list(exc.coord) return body # ─────────────────────────── schemas ─────────────────────────── class RoomResponse(BaseModel): id: int slug: str name: str grid: dict bounds: dict | None spawn: dict | None chunk_count: int class RoomCreate(BaseModel): slug: str = Field(..., min_length=1, max_length=64, pattern=r"^[a-z0-9_-]+$") name: str = Field(..., min_length=1, max_length=255) grid: dict | None = None bounds: dict | None = None spawn: dict | None = None # ─────────────────────────── rooms ─────────────────────────── @router.get("/{world_id}/rooms", response_model=list[RoomResponse]) async def list_rooms(world_id: int, current_user: CurrentUser, db: DbDep): await _get_world_readable(world_id, current_user, db) store = WorldStore(db) rooms = await store.list_rooms(world_id) if not rooms: rooms = [await store.get_or_create_main_room(world_id)] await db.commit() return [ RoomResponse( id=r.id, slug=r.slug, name=r.name, grid=r.grid, bounds=r.bounds, spawn=r.spawn, chunk_count=await store.count_chunks(r.id), ) for r in rooms ] @router.post("/{world_id}/rooms", response_model=RoomResponse, status_code=201) async def create_room(world_id: int, body: RoomCreate, current_user: CurrentUser, db: DbDep): await _get_world_writable(world_id, current_user, db) store = WorldStore(db) if await store.get_room(world_id, body.slug) is not None: raise HTTPException(status_code=409, detail="a room with this slug already exists") room = await store.create_room( world_id, body.slug, body.name, grid=body.grid, bounds=body.bounds, spawn=body.spawn ) await db.commit() await db.refresh(room) return RoomResponse( id=room.id, slug=room.slug, name=room.name, grid=room.grid, bounds=room.bounds, spawn=room.spawn, chunk_count=0, ) # ─────────────────────────── chunk index ─────────────────────────── @router.get("/{world_id}/rooms/{slug}/chunks") async def list_chunks( world_id: int, slug: str, current_user: CurrentUser, db: DbDep, cx_min: int | None = None, cx_max: int | None = None, cy_min: int | None = None, cy_max: int | None = None, cz: int | None = None, ): await _get_world_readable(world_id, current_user, db) store = WorldStore(db) room = await _resolve_room(store, db, world_id, slug) if room is None: raise HTTPException(status_code=404, detail="room not found") if cx_min is not None and cx_max is not None and (cx_max - cx_min + 1) > MAX_BOUNDS_SPAN: raise HTTPException(status_code=422, detail=f"cx bounds span exceeds {MAX_BOUNDS_SPAN}") if cy_min is not None and cy_max is not None and (cy_max - cy_min + 1) > MAX_BOUNDS_SPAN: raise HTTPException(status_code=422, detail=f"cy bounds span exceeds {MAX_BOUNDS_SPAN}") bounds = Bounds(cx_min=cx_min, cx_max=cx_max, cy_min=cy_min, cy_max=cy_max, cz=cz) chunks = await store.list_chunks(world_id, room.id, bounds) content_revision = await store.content_revision(world_id) return { "room": room.slug, "content_revision": content_revision, "chunks": [ { "key": c.chunk_key, "cx": c.cx, "cy": c.cy, "cz": c.cz, "revision": c.revision, "checksum": c.checksum, "payload_len": c.payload_len, "is_mature": c.is_mature, "updated_at": c.updated_at.isoformat() if c.updated_at else None, } for c in chunks ], } # ─────────────────────────── chunk fetch ─────────────────────────── @router.get("/{world_id}/rooms/{slug}/chunks/{cx}/{cy}/{cz}") async def get_chunk( world_id: int, slug: str, cx: int, cy: int, cz: int, current_user: CurrentUser, db: DbDep, accept: Annotated[str | None, Header()] = None, if_none_match: Annotated[str | None, Header(alias="If-None-Match")] = None, ): await _get_world_readable(world_id, current_user, db) store = WorldStore(db) room = await _resolve_room(store, db, world_id, slug) if room is None: raise HTTPException(status_code=404, detail="room not found") key = chunk_codec.build_key(world_id, slug, cx, cy, cz) chunk = await store.get_chunk(key) if chunk is None: raise HTTPException(status_code=404, detail="chunk not found") etag = f'"{chunk.checksum}"' headers = {"ETag": etag, "X-Chunk-Revision": str(chunk.revision), "X-Chunk-Key": key} if if_none_match is not None and if_none_match.strip('"') == chunk.checksum: return Response(status_code=304, headers=headers) wants_json = ( accept is not None and "application/json" in accept and "application/octet-stream" not in accept ) if wants_json: decoded = chunk_codec.decode_binary(chunk.payload) body = chunk_codec.to_json_dict(decoded, key=key, chunk_checksum=chunk.checksum) return JSONResponse(content=body, headers=headers) return Response(content=chunk.payload, media_type="application/octet-stream", headers=headers) # ─────────────────────────── chunk write ─────────────────────────── async def _read_write_payload(request: Request, x_base_revision: int | None) -> tuple[bytes, int]: content_type = request.headers.get("content-type", "") if "application/json" in content_type: body = await request.json() try: payload_bytes = base64.b64decode(body["payload_b64"], validate=True) except (KeyError, binascii.Error, ValueError) as exc: raise HTTPException(status_code=422, detail=f"invalid payload_b64: {exc}") base_revision = int(body.get("base_revision", 0)) return payload_bytes, base_revision payload_bytes = await request.body() return payload_bytes, (x_base_revision if x_base_revision is not None else 0) @router.put("/{world_id}/rooms/{slug}/chunks/{cx}/{cy}/{cz}") async def put_chunk( world_id: int, slug: str, cx: int, cy: int, cz: int, request: Request, current_user: CurrentUser, db: DbDep, x_base_revision: Annotated[int | None, Header(alias="X-Base-Revision")] = None, ): await _get_world_writable(world_id, current_user, db) store = WorldStore(db) room = await _resolve_room(store, db, world_id, slug) if room is None: raise HTTPException(status_code=404, detail="room not found") payload_bytes, base_revision = await _read_write_payload(request, x_base_revision) if len(payload_bytes) > settings.WORLD_CHUNK_MAX_PAYLOAD_BYTES: raise HTTPException( status_code=413, detail=f"payload exceeds WORLD_CHUNK_MAX_PAYLOAD_BYTES ({settings.WORLD_CHUNK_MAX_PAYLOAD_BYTES})", ) try: chunk = await store.put_chunk( room=room, cx=cx, cy=cy, cz=cz, payload_bytes=payload_bytes, base_revision=base_revision, actor_id=current_user.id, ) except chunk_codec.ChunkCodecError as exc: await db.rollback() return JSONResponse(status_code=422, content={"detail": "chunk decode failed", "reason": "malformed", "message": str(exc)}) except chunk_codec.ChunkValidationError as exc: await db.rollback() return JSONResponse(status_code=422, content=_validation_error_body(exc)) except RevisionConflict as exc: # Nothing must stick from a failed write. Returning a JSONResponse # here (rather than raising) means no exception propagates through # `get_db`, so this rollback has to be explicit -- see the note on # the batch endpoint below, where this is load-bearing for # all-or-nothing semantics. await db.rollback() return JSONResponse(status_code=409, content=_conflict_body(exc)) await db.commit() await db.refresh(chunk) return { "key": chunk.chunk_key, "revision": chunk.revision, "checksum": chunk.checksum, "payload_len": chunk.payload_len, "updated_at": chunk.updated_at.isoformat() if chunk.updated_at else None, "is_mature": chunk.is_mature, "live_session": _is_live(world_id), } # ─────────────────────────── batch write ─────────────────────────── class BatchChunkItem(BaseModel): cx: int cy: int cz: int base_revision: int payload_b64: str class BatchPutRequest(BaseModel): chunks: list[BatchChunkItem] @router.put("/{world_id}/rooms/{slug}/chunks") async def put_chunks_batch( world_id: int, slug: str, body: BatchPutRequest, current_user: CurrentUser, db: DbDep, ): await _get_world_writable(world_id, current_user, db) store = WorldStore(db) room = await _resolve_room(store, db, world_id, slug) if room is None: raise HTTPException(status_code=404, detail="room not found") if len(body.chunks) > settings.WORLD_CHUNK_MAX_BATCH: raise HTTPException( status_code=422, detail=f"batch of {len(body.chunks)} exceeds WORLD_CHUNK_MAX_BATCH ({settings.WORLD_CHUNK_MAX_BATCH})", ) items = [] total_bytes = 0 for item in body.chunks: try: payload_bytes = base64.b64decode(item.payload_b64, validate=True) except (binascii.Error, ValueError) as exc: raise HTTPException(status_code=422, detail=f"invalid payload_b64 for ({item.cx},{item.cy},{item.cz}): {exc}") total_bytes += len(payload_bytes) items.append({ "cx": item.cx, "cy": item.cy, "cz": item.cz, "base_revision": item.base_revision, "payload_bytes": payload_bytes, }) if total_bytes > MAX_BATCH_TOTAL_BYTES: raise HTTPException(status_code=413, detail=f"batch exceeds {MAX_BATCH_TOTAL_BYTES} bytes") try: rows = await store.put_chunks_batch(room=room, items=items, actor_id=current_user.id) except chunk_codec.ChunkCodecError as exc: # All-or-nothing (world_persistence_impl.md §8.1): earlier items in # this batch may already be flushed (but never committed) to this # transaction. Rolling back here is what actually makes the batch # atomic -- without it, returning a JSONResponse instead of raising # means no exception reaches `get_db`, and its own transaction # handling would otherwise commit those earlier writes anyway. await db.rollback() return JSONResponse(status_code=422, content={"detail": "chunk decode failed", "reason": "malformed", "message": str(exc)}) except chunk_codec.ChunkValidationError as exc: await db.rollback() return JSONResponse(status_code=422, content=_validation_error_body(exc)) except RevisionConflict as exc: await db.rollback() return JSONResponse(status_code=409, content=_conflict_body(exc)) await db.commit() for row in rows: await db.refresh(row) return { "live_session": _is_live(world_id), "chunks": [ { "key": r.chunk_key, "revision": r.revision, "checksum": r.checksum, "payload_len": r.payload_len, "updated_at": r.updated_at.isoformat() if r.updated_at else None, "is_mature": r.is_mature, } for r in rows ], } # ─────────────────────────── delete ─────────────────────────── @router.delete("/{world_id}/rooms/{slug}/chunks/{cx}/{cy}/{cz}", status_code=204) async def delete_chunk( world_id: int, slug: str, cx: int, cy: int, cz: int, current_user: CurrentUser, db: DbDep, ): await _get_world_writable(world_id, current_user, db) store = WorldStore(db) room = await _resolve_room(store, db, world_id, slug) if room is None: raise HTTPException(status_code=404, detail="room not found") key = chunk_codec.build_key(world_id, slug, cx, cy, cz) await store.delete_chunk(key, current_user.id) await db.commit() return Response(status_code=204) # ─────────────────────────── revisions + rollback ─────────────────────────── @router.get("/{world_id}/rooms/{slug}/chunks/{cx}/{cy}/{cz}/revisions") async def chunk_revisions( world_id: int, slug: str, cx: int, cy: int, cz: int, current_user: CurrentUser, db: DbDep, ): # doc: "(owner)" -- same gate as writes. await _get_world_writable(world_id, current_user, db) store = WorldStore(db) key = chunk_codec.build_key(world_id, slug, cx, cy, cz) revisions = await store.list_revisions(key) return [ { "revision": r.revision, "action": r.action, "actor_id": r.actor_id, "checksum": r.checksum, "payload_available": r.payload is not None, "created_at": r.created_at.isoformat() if r.created_at else None, } for r in revisions ] class RollbackRequest(BaseModel): revision: int @router.post("/{world_id}/rooms/{slug}/chunks/{cx}/{cy}/{cz}/rollback") async def rollback_chunk( world_id: int, slug: str, cx: int, cy: int, cz: int, body: RollbackRequest, current_user: CurrentUser, db: DbDep, ): await _get_world_writable(world_id, current_user, db) store = WorldStore(db) key = chunk_codec.build_key(world_id, slug, cx, cy, cz) try: chunk = await store.rollback_chunk(key, body.revision, current_user.id) except RevisionNotAvailable as exc: await db.rollback() return JSONResponse(status_code=404, content={"detail": str(exc)}) except chunk_codec.ChunkValidationError as exc: await db.rollback() return JSONResponse(status_code=422, content=_validation_error_body(exc)) await db.commit() await db.refresh(chunk) return { "key": chunk.chunk_key, "revision": chunk.revision, "checksum": chunk.checksum, "payload_len": chunk.payload_len, "updated_at": chunk.updated_at.isoformat() if chunk.updated_at else None, "is_mature": chunk.is_mature, "live_session": _is_live(world_id), }