feat: infer genlocke visibility from first leg's run

Genlockes now inherit visibility from their first leg's run:
- Private runs make genlockes hidden from public listings
- All genlocke read endpoints now accept optional auth
- Returns 404 for private genlockes to non-owners

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-21 13:47:05 +01:00
parent a3f332f82b
commit a4fa5bf1e4
3 changed files with 383 additions and 41 deletions

View File

@@ -8,14 +8,14 @@ from sqlalchemy import update as sa_update
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
from app.core.auth import AuthUser, require_auth, require_run_owner
from app.core.auth import AuthUser, get_current_user, require_auth, require_run_owner
from app.core.database import get_session
from app.models.encounter import Encounter
from app.models.evolution import Evolution
from app.models.game import Game
from app.models.genlocke import Genlocke, GenlockeLeg
from app.models.genlocke_transfer import GenlockeTransfer
from app.models.nuzlocke_run import NuzlockeRun
from app.models.nuzlocke_run import NuzlockeRun, RunVisibility
from app.models.pokemon import Pokemon
from app.models.route import Route
from app.models.user import User
@@ -77,8 +77,34 @@ async def _check_genlocke_owner(
require_run_owner(first_leg.run, user)
def _is_genlocke_visible(genlocke: Genlocke, user: AuthUser | None) -> bool:
"""
Check if a genlocke is visible to the given user.
Visibility is inferred from the first leg's run:
- Public runs are visible to everyone
- Private runs are only visible to the owner
"""
first_leg = next((leg for leg in genlocke.legs if leg.leg_order == 1), None)
if not first_leg or not first_leg.run:
# No first leg or run - treat as visible (legacy data)
return True
if first_leg.run.visibility == RunVisibility.PUBLIC:
return True
# Private run - only visible to owner
if user is None:
return False
if first_leg.run.owner_id is None:
return False
return str(first_leg.run.owner_id) == user.id
@router.get("", response_model=list[GenlockeListItem])
async def list_genlockes(session: AsyncSession = Depends(get_session)):
async def list_genlockes(
session: AsyncSession = Depends(get_session),
user: AuthUser | None = Depends(get_current_user),
):
result = await session.execute(
select(Genlocke)
.options(
@@ -92,6 +118,10 @@ async def list_genlockes(session: AsyncSession = Depends(get_session)):
items = []
for g in genlockes:
# Filter out private genlockes for non-owners
if not _is_genlocke_visible(g, user):
continue
completed_legs = 0
current_leg_order = None
owner = None
@@ -126,7 +156,11 @@ async def list_genlockes(session: AsyncSession = Depends(get_session)):
@router.get("/{genlocke_id}", response_model=GenlockeDetailResponse)
async def get_genlocke(genlocke_id: int, session: AsyncSession = Depends(get_session)):
async def get_genlocke(
genlocke_id: int,
session: AsyncSession = Depends(get_session),
user: AuthUser | None = Depends(get_current_user),
):
result = await session.execute(
select(Genlocke)
.where(Genlocke.id == genlocke_id)
@@ -139,6 +173,10 @@ async def get_genlocke(genlocke_id: int, session: AsyncSession = Depends(get_ses
if genlocke is None:
raise HTTPException(status_code=404, detail="Genlocke not found")
# Check visibility - return 404 for private genlockes to non-owners
if not _is_genlocke_visible(genlocke, user):
raise HTTPException(status_code=404, detail="Genlocke not found")
# Collect run IDs for aggregate query
run_ids = [leg.run_id for leg in genlocke.legs if leg.run_id is not None]
@@ -222,20 +260,26 @@ async def get_genlocke(genlocke_id: int, session: AsyncSession = Depends(get_ses
response_model=GenlockeGraveyardResponse,
)
async def get_genlocke_graveyard(
genlocke_id: int, session: AsyncSession = Depends(get_session)
genlocke_id: int,
session: AsyncSession = Depends(get_session),
user: AuthUser | None = Depends(get_current_user),
):
# Load genlocke with legs + game
# Load genlocke with legs + game + run (for visibility check)
result = await session.execute(
select(Genlocke)
.where(Genlocke.id == genlocke_id)
.options(
selectinload(Genlocke.legs).selectinload(GenlockeLeg.game),
selectinload(Genlocke.legs).selectinload(GenlockeLeg.run),
)
)
genlocke = result.scalar_one_or_none()
if genlocke is None:
raise HTTPException(status_code=404, detail="Genlocke not found")
if not _is_genlocke_visible(genlocke, user):
raise HTTPException(status_code=404, detail="Genlocke not found")
# Build run_id → (leg_order, game_name) lookup
run_ids = [leg.run_id for leg in genlocke.legs if leg.run_id is not None]
run_lookup: dict[int, tuple[int, str]] = {}
@@ -323,20 +367,26 @@ async def get_genlocke_graveyard(
response_model=GenlockeLineageResponse,
)
async def get_genlocke_lineages(
genlocke_id: int, session: AsyncSession = Depends(get_session)
genlocke_id: int,
session: AsyncSession = Depends(get_session),
user: AuthUser | None = Depends(get_current_user),
):
# Load genlocke with legs + game
# Load genlocke with legs + game + run (for visibility check)
result = await session.execute(
select(Genlocke)
.where(Genlocke.id == genlocke_id)
.options(
selectinload(Genlocke.legs).selectinload(GenlockeLeg.game),
selectinload(Genlocke.legs).selectinload(GenlockeLeg.run),
)
)
genlocke = result.scalar_one_or_none()
if genlocke is None:
raise HTTPException(status_code=404, detail="Genlocke not found")
if not _is_genlocke_visible(genlocke, user):
raise HTTPException(status_code=404, detail="Genlocke not found")
# Query all transfers for this genlocke
transfer_result = await session.execute(
select(GenlockeTransfer).where(GenlockeTransfer.genlocke_id == genlocke_id)
@@ -570,15 +620,23 @@ async def get_leg_survivors(
genlocke_id: int,
leg_order: int,
session: AsyncSession = Depends(get_session),
user: AuthUser | None = Depends(get_current_user),
):
# Find the leg
result = await session.execute(
select(GenlockeLeg).where(
GenlockeLeg.genlocke_id == genlocke_id,
GenlockeLeg.leg_order == leg_order,
)
# Load genlocke with legs + run for visibility check
genlocke_result = await session.execute(
select(Genlocke)
.where(Genlocke.id == genlocke_id)
.options(selectinload(Genlocke.legs).selectinload(GenlockeLeg.run))
)
leg = result.scalar_one_or_none()
genlocke = genlocke_result.scalar_one_or_none()
if genlocke is None:
raise HTTPException(status_code=404, detail="Genlocke not found")
if not _is_genlocke_visible(genlocke, user):
raise HTTPException(status_code=404, detail="Genlocke not found")
# Find the leg
leg = next((leg for leg in genlocke.legs if leg.leg_order == leg_order), None)
if leg is None:
raise HTTPException(status_code=404, detail="Leg not found")
@@ -846,12 +904,21 @@ class RetiredFamiliesResponse(BaseModel):
async def get_retired_families(
genlocke_id: int,
session: AsyncSession = Depends(get_session),
user: AuthUser | None = Depends(get_current_user),
):
# Verify genlocke exists
genlocke = await session.get(Genlocke, genlocke_id)
# Load genlocke with legs + run for visibility check
result = await session.execute(
select(Genlocke)
.where(Genlocke.id == genlocke_id)
.options(selectinload(Genlocke.legs).selectinload(GenlockeLeg.run))
)
genlocke = result.scalar_one_or_none()
if genlocke is None:
raise HTTPException(status_code=404, detail="Genlocke not found")
if not _is_genlocke_visible(genlocke, user):
raise HTTPException(status_code=404, detail="Genlocke not found")
# Query all legs with retired_pokemon_ids
result = await session.execute(
select(GenlockeLeg)