Add lobbies: DM sees all characters, players only a short card of others
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HJhXuLD2Z6hv3Wqeq9UL9K
This commit is contained in:
Claude
+184
-2
@@ -22,6 +22,9 @@ MAX_BODY = 200_000 # bytes per character
|
||||
MAX_PRESETS = 200
|
||||
MAX_CHARACTERS = 300 # per user
|
||||
MAX_SHARES = 30 # per character
|
||||
MAX_LOBBIES = 20 # per DM
|
||||
MAX_MEMBERS = 20 # per lobby
|
||||
CODE_ALPHABET = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789" # no 0/O/1/I
|
||||
|
||||
# No public API docs: everything about the API stays behind the login
|
||||
app = FastAPI(title="Arcane Wasteland Character Creator", docs_url=None, redoc_url=None, openapi_url=None)
|
||||
@@ -48,6 +51,11 @@ def db() -> sqlite3.Connection:
|
||||
auth.init_tables(con)
|
||||
con.execute("""CREATE TABLE IF NOT EXISTS shares (
|
||||
character_id TEXT NOT NULL, user_id INTEGER NOT NULL, PRIMARY KEY (character_id, user_id))""")
|
||||
con.execute("""CREATE TABLE IF NOT EXISTS lobbies (
|
||||
id TEXT PRIMARY KEY, name TEXT NOT NULL, owner_id INTEGER NOT NULL, code TEXT NOT NULL UNIQUE, created REAL NOT NULL)""")
|
||||
con.execute("""CREATE TABLE IF NOT EXISTS lobby_members (
|
||||
lobby_id TEXT NOT NULL, user_id INTEGER NOT NULL, character_id TEXT, joined REAL NOT NULL,
|
||||
PRIMARY KEY (lobby_id, user_id))""")
|
||||
for table in ("characters", "presets"): # older databases: add the owner column
|
||||
if "owner_id" not in [r["name"] for r in con.execute(f"PRAGMA table_info({table})")]:
|
||||
con.execute(f"ALTER TABLE {table} ADD COLUMN owner_id INTEGER")
|
||||
@@ -326,11 +334,13 @@ def create_character(ch: dict = Body(...), user: dict = Depends(current_user)):
|
||||
|
||||
@api.get("/characters/{cid}")
|
||||
def read_character(cid: str, user: dict = Depends(current_user)):
|
||||
"""Only the owner and the users the character is shared with may read it."""
|
||||
"""Readable by the owner, by users the character is shared with, and by the DM of a lobby it plays in."""
|
||||
with db() as con:
|
||||
row = con.execute("SELECT data, owner_id FROM characters WHERE id=?", (cid,)).fetchone()
|
||||
allowed = bool(row) and (row["owner_id"] == user["id"] or con.execute(
|
||||
"SELECT 1 FROM shares WHERE character_id=? AND user_id=?", (cid, user["id"])).fetchone())
|
||||
"SELECT 1 FROM shares WHERE character_id=? AND user_id=?", (cid, user["id"])).fetchone() or con.execute(
|
||||
"""SELECT 1 FROM lobby_members m JOIN lobbies l ON l.id = m.lobby_id
|
||||
WHERE m.character_id=? AND l.owner_id=?""", (cid, user["id"])).fetchone())
|
||||
if not allowed:
|
||||
raise HTTPException(404, "Character not found")
|
||||
return json.loads(row["data"])
|
||||
@@ -376,6 +386,178 @@ def delete_character(cid: str, user: dict = Depends(current_user)):
|
||||
_owned(con, "characters", cid, user)
|
||||
con.execute("DELETE FROM characters WHERE id=?", (cid,))
|
||||
con.execute("DELETE FROM shares WHERE character_id=?", (cid,))
|
||||
con.execute("UPDATE lobby_members SET character_id=NULL WHERE character_id=?", (cid,))
|
||||
return {"ok": True}
|
||||
|
||||
|
||||
# ---- lobbies: the creator is the DM and sees every member's character; players only see a short card of the others
|
||||
|
||||
def _lobby_for(con, lid: str, user: dict):
|
||||
"""Returns the lobby row if the user is its DM or a member, otherwise 404 (hides that it exists)."""
|
||||
row = con.execute("""SELECT l.*, u.username AS dm FROM lobbies l JOIN users u ON u.id = l.owner_id WHERE l.id=?""",
|
||||
(lid,)).fetchone()
|
||||
if not row or (row["owner_id"] != user["id"] and not con.execute(
|
||||
"SELECT 1 FROM lobby_members WHERE lobby_id=? AND user_id=?", (lid, user["id"])).fetchone()):
|
||||
raise HTTPException(404, "Lobby not found")
|
||||
return row
|
||||
|
||||
|
||||
def _dm_only(lobby, user: dict):
|
||||
if lobby["owner_id"] != user["id"]:
|
||||
raise HTTPException(403, "Only the DM can do this")
|
||||
|
||||
|
||||
def _new_code(con) -> str:
|
||||
while True:
|
||||
code = "".join(secrets.choice(CODE_ALPHABET) for _ in range(8))
|
||||
if not con.execute("SELECT 1 FROM lobbies WHERE code=?", (code,)).fetchone():
|
||||
return code
|
||||
|
||||
|
||||
def _own_character(con, cid, user: dict):
|
||||
row = con.execute("SELECT id FROM characters WHERE id=? AND owner_id=?", (cid, user["id"])).fetchone()
|
||||
if not row:
|
||||
raise HTTPException(422, "Pick one of your own characters")
|
||||
|
||||
|
||||
def _card(data: str, rules: dict) -> dict:
|
||||
"""The superficial look at a character that fellow players get: name, concept and class/profession names only."""
|
||||
ch = json.loads(data)
|
||||
cls = next((c for c in rules["classes"] if c["id"] == ch.get("class_id")), None)
|
||||
sub = next((x for x in (cls or {}).get("subclasses", []) if x["id"] == ch.get("subclass_id")), None)
|
||||
prof = next((p for p in rules["professions"] if p["id"] == ch.get("profession")), None)
|
||||
return {"name": ch.get("name") or "Nameless", "concept": str(ch.get("concept") or "")[:120],
|
||||
"class": (cls or {}).get("name", ""), "subclass": (sub or {}).get("name", ""),
|
||||
"profession": (prof or {}).get("name", "")}
|
||||
|
||||
|
||||
@api.get("/lobbies")
|
||||
def my_lobbies(user: dict = Depends(current_user)):
|
||||
with db() as con:
|
||||
mine = con.execute("""SELECT l.id, l.name, (SELECT COUNT(*) FROM lobby_members m WHERE m.lobby_id = l.id) AS n
|
||||
FROM lobbies l WHERE l.owner_id=? ORDER BY l.created DESC""", (user["id"],)).fetchall()
|
||||
joined = con.execute("""SELECT l.id, l.name, u.username AS dm, c.data FROM lobby_members m
|
||||
JOIN lobbies l ON l.id = m.lobby_id JOIN users u ON u.id = l.owner_id
|
||||
LEFT JOIN characters c ON c.id = m.character_id
|
||||
WHERE m.user_id=? ORDER BY m.joined DESC""", (user["id"],)).fetchall()
|
||||
return {"mine": [{"id": r["id"], "name": r["name"], "members": r["n"]} for r in mine],
|
||||
"joined": [{"id": r["id"], "name": r["name"], "dm": r["dm"], "character": _name(r["data"]) if r["data"] else ""}
|
||||
for r in joined]}
|
||||
|
||||
|
||||
@api.post("/lobbies")
|
||||
def create_lobby(body: dict = Body(...), user: dict = Depends(current_user)):
|
||||
name = str(body.get("name", "")).strip()[:60]
|
||||
if not name:
|
||||
raise HTTPException(422, "Give the lobby a name")
|
||||
with db() as con:
|
||||
if con.execute("SELECT COUNT(*) FROM lobbies WHERE owner_id=?", (user["id"],)).fetchone()[0] >= MAX_LOBBIES:
|
||||
raise HTTPException(409, "Lobby limit reached")
|
||||
lid, code = uuid.uuid4().hex[:12], _new_code(con)
|
||||
con.execute("INSERT INTO lobbies (id, name, owner_id, code, created) VALUES (?,?,?,?,?)",
|
||||
(lid, name, user["id"], code, time.time()))
|
||||
return {"id": lid, "code": code}
|
||||
|
||||
|
||||
@api.post("/lobbies/join")
|
||||
def join_lobby(request: Request, body: dict = Body(...), user: dict = Depends(current_user)):
|
||||
ip = client_ip(request)
|
||||
throttle(f"join:{ip}", 10, 600) # only wrong codes count, so guessing is slow
|
||||
throttle(f"join:u{user['id']}", 10, 600)
|
||||
code = re.sub(r"[^A-Za-z0-9]", "", str(body.get("code", ""))).upper()
|
||||
with db() as con:
|
||||
lobby = con.execute("SELECT * FROM lobbies WHERE code=?", (code,)).fetchone()
|
||||
if not lobby:
|
||||
limiter.fail(f"join:{ip}"); limiter.fail(f"join:u{user['id']}")
|
||||
raise HTTPException(404, "No lobby with this code")
|
||||
if lobby["owner_id"] == user["id"]:
|
||||
raise HTTPException(422, "You are the DM of this lobby")
|
||||
_own_character(con, body.get("character_id"), user)
|
||||
known = con.execute("SELECT 1 FROM lobby_members WHERE lobby_id=? AND user_id=?", (lobby["id"], user["id"])).fetchone()
|
||||
if not known and con.execute("SELECT COUNT(*) FROM lobby_members WHERE lobby_id=?", (lobby["id"],)).fetchone()[0] >= MAX_MEMBERS:
|
||||
raise HTTPException(409, "This lobby is full")
|
||||
con.execute("""INSERT INTO lobby_members (lobby_id, user_id, character_id, joined) VALUES (?,?,?,?)
|
||||
ON CONFLICT(lobby_id, user_id) DO UPDATE SET character_id=excluded.character_id""",
|
||||
(lobby["id"], user["id"], body["character_id"], time.time()))
|
||||
return {"id": lobby["id"]}
|
||||
|
||||
|
||||
@api.get("/lobbies/{lid}")
|
||||
def read_lobby(lid: str, user: dict = Depends(current_user)):
|
||||
rules = load_rules()
|
||||
with db() as con:
|
||||
lobby = _lobby_for(con, lid, user)
|
||||
is_dm = lobby["owner_id"] == user["id"]
|
||||
rows = con.execute("""SELECT m.user_id, m.character_id, u.username, c.data FROM lobby_members m
|
||||
JOIN users u ON u.id = m.user_id LEFT JOIN characters c ON c.id = m.character_id
|
||||
WHERE m.lobby_id=? ORDER BY m.joined""", (lid,)).fetchall()
|
||||
members = []
|
||||
for r in rows:
|
||||
mine = r["user_id"] == user["id"]
|
||||
card = _card(r["data"], rules) if r["data"] else None
|
||||
# The id is only given out where the viewer may open the full character (the DM sees all, a player only their own)
|
||||
members.append({"username": r["username"], "me": mine, "character": card,
|
||||
"character_id": r["character_id"] if (is_dm or mine) and card else None})
|
||||
return {"id": lobby["id"], "name": lobby["name"], "dm": lobby["dm"], "is_dm": is_dm,
|
||||
"code": lobby["code"] if is_dm else None, "members": members}
|
||||
|
||||
|
||||
@api.put("/lobbies/{lid}")
|
||||
def rename_lobby(lid: str, body: dict = Body(...), user: dict = Depends(current_user)):
|
||||
name = str(body.get("name", "")).strip()[:60]
|
||||
if not name:
|
||||
raise HTTPException(422, "Give the lobby a name")
|
||||
with db() as con:
|
||||
_dm_only(_lobby_for(con, lid, user), user)
|
||||
con.execute("UPDATE lobbies SET name=? WHERE id=?", (name, lid))
|
||||
return {"ok": True}
|
||||
|
||||
|
||||
@api.delete("/lobbies/{lid}")
|
||||
def delete_lobby(lid: str, user: dict = Depends(current_user)):
|
||||
with db() as con:
|
||||
_dm_only(_lobby_for(con, lid, user), user)
|
||||
con.execute("DELETE FROM lobby_members WHERE lobby_id=?", (lid,))
|
||||
con.execute("DELETE FROM lobbies WHERE id=?", (lid,))
|
||||
return {"ok": True}
|
||||
|
||||
|
||||
@api.post("/lobbies/{lid}/code")
|
||||
def new_lobby_code(lid: str, user: dict = Depends(current_user)):
|
||||
"""New join code, the old one stops working (members stay)."""
|
||||
with db() as con:
|
||||
_dm_only(_lobby_for(con, lid, user), user)
|
||||
code = _new_code(con)
|
||||
con.execute("UPDATE lobbies SET code=? WHERE id=?", (code, lid))
|
||||
return {"code": code}
|
||||
|
||||
|
||||
@api.put("/lobbies/{lid}/me")
|
||||
def switch_lobby_character(lid: str, body: dict = Body(...), user: dict = Depends(current_user)):
|
||||
with db() as con:
|
||||
_lobby_for(con, lid, user)
|
||||
_own_character(con, body.get("character_id"), user)
|
||||
cur = con.execute("UPDATE lobby_members SET character_id=? WHERE lobby_id=? AND user_id=?",
|
||||
(body["character_id"], lid, user["id"]))
|
||||
if not cur.rowcount:
|
||||
raise HTTPException(422, "You are the DM of this lobby")
|
||||
return {"ok": True}
|
||||
|
||||
|
||||
@api.delete("/lobbies/{lid}/me")
|
||||
def leave_lobby(lid: str, user: dict = Depends(current_user)):
|
||||
with db() as con:
|
||||
_lobby_for(con, lid, user)
|
||||
con.execute("DELETE FROM lobby_members WHERE lobby_id=? AND user_id=?", (lid, user["id"]))
|
||||
return {"ok": True}
|
||||
|
||||
|
||||
@api.delete("/lobbies/{lid}/members/{username}")
|
||||
def kick_member(lid: str, username: str, user: dict = Depends(current_user)):
|
||||
with db() as con:
|
||||
_dm_only(_lobby_for(con, lid, user), user)
|
||||
con.execute("DELETE FROM lobby_members WHERE lobby_id=? AND user_id=(SELECT id FROM users WHERE username=?)",
|
||||
(lid, username))
|
||||
return {"ok": True}
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user