AWCharacterCreator/backend/app/maps.py

"""Maps for the lobby table: the DM's map library, the active map of a lobby and the tokens on it.

Coordinates are pixels of the stored image. A token's size is in grid squares (the map's `grid` is pixels per square).
Images are re-encoded with Pillow on upload, so only real images get through and metadata is gone."""
import hashlib
import io
import json
import shutil
import time
import uuid
from pathlib import Path

from fastapi import Body, Depends, HTTPException, Query, Request
from fastapi.concurrency import run_in_threadpool
from fastapi.responses import Response
from PIL import Image, ImageOps

from . import vision
from .portrait import tag as portrait_tag

MAX_UPLOAD = 12_000_000     # bytes of the uploaded file
MAX_SIDE = 4096             # larger images are scaled down
MAX_PIXELS = 60_000_000     # decompression bomb guard (before scaling)
MAX_MAPS = 30               # per DM
MAX_TOKENS = 120            # per map / lobby
MAX_WALLS = 400             # wall and door segments per map
MAX_SHARES = 30
COLORS = ("#c0392b", "#8e44ad", "#16a085", "#7f8c8d", "#d35400", "#2c3e50", "#27ae60", "#b9770e")


def schema(con):
    con.execute("""CREATE TABLE IF NOT EXISTS maps (
        id TEXT PRIMARY KEY, owner_id INTEGER NOT NULL, name TEXT NOT NULL, w INTEGER NOT NULL, h INTEGER NOT NULL,
        grid REAL NOT NULL, tokens TEXT NOT NULL DEFAULT '[]', walls TEXT NOT NULL DEFAULT '[]', created REAL NOT NULL)""")
    con.execute("""CREATE TABLE IF NOT EXISTS lobby_tokens (
        id TEXT PRIMARY KEY, lobby_id TEXT NOT NULL, user_id INTEGER, name TEXT NOT NULL, x REAL, y REAL,
        color TEXT, size REAL NOT NULL DEFAULT 1, hidden INTEGER NOT NULL DEFAULT 0)""")
    con.execute("CREATE INDEX IF NOT EXISTS lobby_tokens_lobby ON lobby_tokens (lobby_id)")
    con.execute("CREATE TABLE IF NOT EXISTS map_shares (map_id TEXT NOT NULL, user_id INTEGER NOT NULL, PRIMARY KEY (map_id, user_id))")
    for table, col, ddl in (("lobbies", "map_id", "TEXT"), ("lobbies", "doors", "TEXT NOT NULL DEFAULT '{}'"),
                            ("lobbies", "los", "INTEGER NOT NULL DEFAULT 1"), ("maps", "sight", "REAL NOT NULL DEFAULT 0")):
        if col not in [r["name"] for r in con.execute(f"PRAGMA table_info({table})")]:
            con.execute(f"ALTER TABLE {table} ADD COLUMN {col} {ddl}")


def encode(raw: bytes) -> tuple[bytes, int, int]:
    """Validates and re-encodes an uploaded image as WebP, scaled to at most MAX_SIDE."""
    Image.MAX_IMAGE_PIXELS = MAX_PIXELS
    try:
        im = Image.open(io.BytesIO(raw))
        if im.format not in ("PNG", "JPEG", "WEBP"):
            raise ValueError("format")
        im.load()
    except Exception:
        raise HTTPException(422, "Please upload a PNG, JPEG or WebP image")
    im = ImageOps.exif_transpose(im)
    im.thumbnail((MAX_SIDE, MAX_SIDE))
    if im.mode not in ("RGB", "RGBA"):
        im = im.convert("RGBA" if "transparency" in im.info or im.mode in ("LA", "PA") else "RGB")
    out = io.BytesIO()
    im.save(out, "WEBP", quality=90, method=4)
    return out.getvalue(), im.width, im.height


def clean_color(c, default="#c0392b") -> str:
    c = str(c or "")
    return c if len(c) == 7 and c[0] == "#" and all(ch in "0123456789abcdefABCDEF" for ch in c[1:]) else default


def clean_token(t: dict, w: int, h: int, keep_id=True) -> dict:
    if not isinstance(t, dict):
        raise HTTPException(422, "Invalid token")
    name = str(t.get("name", "")).strip()[:40] or "Token"
    try:
        x, y = t.get("x"), t.get("y")
        x = None if x is None else min(max(float(x), 0), w)
        y = None if y is None else min(max(float(y), 0), h)
        size = min(max(float(t.get("size", 1)), 0.25), 8)
    except (TypeError, ValueError):
        raise HTTPException(422, "Invalid token position or size")
    if (x is None) != (y is None):
        x = y = None
    tid = str(t.get("id", "")) if keep_id else ""
    return {"id": tid if tid.isalnum() and len(tid) <= 16 else uuid.uuid4().hex[:12], "name": name, "x": x, "y": y,
            "size": size, "color": clean_color(t.get("color")), "hidden": bool(t.get("hidden"))}


def clean_walls(raw, w: int, h: int) -> list:
    if not isinstance(raw, list) or len(raw) > MAX_WALLS:
        raise HTTPException(422, f"At most {MAX_WALLS} wall segments per map")
    out = []
    for t in raw:
        try:
            x1, y1, x2, y2 = (float(t[k]) for k in ("x1", "y1", "x2", "y2"))
        except (TypeError, ValueError, KeyError, IndexError):
            raise HTTPException(422, "Invalid wall")
        x1, x2 = min(max(x1, 0), w), min(max(x2, 0), w)
        y1, y2 = min(max(y1, 0), h), min(max(y2, 0), h)
        if (x1, y1) == (x2, y2):
            continue
        wid = str(t.get("id", ""))
        out.append({"id": wid if wid.isalnum() and len(wid) <= 16 else uuid.uuid4().hex[:10],
                    "x1": round(x1, 1), "y1": round(y1, 1), "x2": round(x2, 1), "y2": round(y2, 1),
                    "kind": "door" if t.get("kind") == "door" else "wall"})
    return out


def install(api, db, current_user, lobby_for, dm_only, hub, event, store_dir: Path):
    store_dir.mkdir(parents=True, exist_ok=True)

    def path(mid: str) -> Path:
        return store_dir / f"{mid}.webp"

    def own_map(con, mid, user):
        row = con.execute("SELECT * FROM maps WHERE id=? AND owner_id=?", (mid, user["id"])).fetchone()
        if not row:
            raise HTTPException(404, "Map not found")
        return row

    def usable_map(con, mid, user):
        """A map the user owns or that was shared with them (shared maps can be put on the table but not changed)."""
        row = con.execute("""SELECT * FROM maps WHERE id=? AND (owner_id=? OR EXISTS
                             (SELECT 1 FROM map_shares s WHERE s.map_id = maps.id AND s.user_id=?))""", (mid, user["id"], user["id"])).fetchone()
        if not row:
            raise HTTPException(404, "Map not found")
        return row

    def meta(r) -> dict:
        return {"id": r["id"], "name": r["name"], "w": r["w"], "h": r["h"], "grid": r["grid"], "sight": r["sight"]}

    # ---- the DM's library

    @api.get("/maps")
    def list_maps(user: dict = Depends(current_user)):
        with db() as con:
            rows = con.execute("""SELECT m.*, u.username AS author FROM maps m JOIN users u ON u.id = m.owner_id
                                  WHERE m.owner_id=? OR EXISTS (SELECT 1 FROM map_shares s WHERE s.map_id = m.id AND s.user_id=?)
                                  ORDER BY m.created DESC""", (user["id"], user["id"])).fetchall()
            out = []
            for r in rows:
                mine = r["owner_id"] == user["id"]
                shared = [x["username"] for x in con.execute(
                    "SELECT u.username FROM map_shares s JOIN users u ON u.id = s.user_id WHERE s.map_id=? ORDER BY u.username", (r["id"],))] if mine else []
                out.append({**meta(r), "tokens": len(json.loads(r["tokens"])), "walls": len(json.loads(r["walls"])),
                            "mine": mine, "author": r["author"], "shared_with": shared})
        return out

    @api.post("/maps")
    async def upload_map(request: Request, name: str = "", user: dict = Depends(current_user)):
        """Raw image bytes in the body (no multipart), the name as ?name=."""
        if int(request.headers.get("content-length") or 0) > MAX_UPLOAD:
            raise HTTPException(413, f"The file is too large (at most {MAX_UPLOAD // 1_000_000} MB)")
        chunks, size = [], 0
        async for c in request.stream():
            size += len(c)
            if size > MAX_UPLOAD:
                raise HTTPException(413, f"The file is too large (at most {MAX_UPLOAD // 1_000_000} MB)")
            chunks.append(c)
        with db() as con:
            if con.execute("SELECT COUNT(*) FROM maps WHERE owner_id=?", (user["id"],)).fetchone()[0] >= MAX_MAPS:
                raise HTTPException(409, "Map limit reached, delete one first")
        data, w, h = await run_in_threadpool(encode, b"".join(chunks))
        mid = uuid.uuid4().hex[:12]
        path(mid).write_bytes(data)
        with db() as con:
            con.execute("INSERT INTO maps (id, owner_id, name, w, h, grid, created) VALUES (?,?,?,?,?,?,?)",
                        (mid, user["id"], name.strip()[:60] or "Map", w, h, round(max(w, h) / 30, 1), time.time()))
        return {"id": mid}

    @api.get("/maps/{mid}")
    def read_map(mid: str, user: dict = Depends(current_user)):
        with db() as con:
            r = own_map(con, mid, user)
        return {**meta(r), "tokens": json.loads(r["tokens"]), "walls": json.loads(r["walls"])}

    @api.put("/maps/{mid}")
    def update_map(mid: str, body: dict = Body(...), user: dict = Depends(current_user)):
        with db() as con:
            r = own_map(con, mid, user)
            name, grid, tokens, walls, sight = r["name"], r["grid"], r["tokens"], r["walls"], r["sight"]
            if "name" in body:
                name = str(body["name"]).strip()[:60] or r["name"]
            if "grid" in body:
                try:
                    grid = float(body["grid"])
                except (TypeError, ValueError):
                    raise HTTPException(422, "The grid must be a number")
                if not 4 <= grid <= max(r["w"], r["h"]) / 2:
                    raise HTTPException(422, "Grid size (pixels per square) is out of range")
            if "tokens" in body:
                if not isinstance(body["tokens"], list) or len(body["tokens"]) > MAX_TOKENS:
                    raise HTTPException(422, f"At most {MAX_TOKENS} tokens per map")
                tokens = json.dumps([clean_token(t, r["w"], r["h"]) for t in body["tokens"]])
            if "walls" in body:
                walls = json.dumps(clean_walls(body["walls"], r["w"], r["h"]))
            if "sight" in body:
                try:
                    sight = float(body["sight"])
                except (TypeError, ValueError):
                    raise HTTPException(422, "The sight range must be a number")
                if not 0 <= sight <= 1000:
                    raise HTTPException(422, "Sight range: 0 (unlimited) to 1000 squares")
            con.execute("UPDATE maps SET name=?, grid=?, tokens=?, walls=?, sight=? WHERE id=?", (name, grid, tokens, walls, sight, mid))
            lobbies = [x["id"] for x in con.execute("SELECT id FROM lobbies WHERE map_id=?", (mid,))]
        for lid in lobbies:
            hub.notify(lid, "map")
        return {"ok": True}

    @api.delete("/maps/{mid}")
    def delete_map(mid: str, user: dict = Depends(current_user)):
        with db() as con:
            own_map(con, mid, user)
            lobbies = [x["id"] for x in con.execute("SELECT id FROM lobbies WHERE map_id=?", (mid,))]
            con.execute("UPDATE lobbies SET map_id=NULL WHERE map_id=?", (mid,))
            con.execute("DELETE FROM map_shares WHERE map_id=?", (mid,))
            for lid in lobbies:
                con.execute("DELETE FROM lobby_tokens WHERE lobby_id=?", (lid,))
            con.execute("DELETE FROM maps WHERE id=?", (mid,))
        path(mid).unlink(missing_ok=True)
        (store_dir / f"{mid}.thumb.webp").unlink(missing_ok=True)
        for lid in lobbies:
            hub.notify(lid, "map")
        return {"ok": True}

    @api.get("/maps/{mid}/image")
    def map_image(mid: str, user: dict = Depends(current_user)):
        with db() as con:
            own_map(con, mid, user)
        return image_response(path(mid))

    @api.get("/maps/{mid}/thumb")
    def map_thumb(mid: str, user: dict = Depends(current_user)):
        with db() as con:
            usable_map(con, mid, user)
        t = store_dir / f"{mid}.thumb.webp"
        if not t.exists():
            im = Image.open(path(mid)).convert("RGB")
            im.thumbnail((320, 320))
            im.save(t, "WEBP", quality=80)
        return image_response(t)

    @api.post("/maps/{mid}/shares")
    def share_map(mid: str, body: dict = Body(...), user: dict = Depends(current_user)):
        name = str(body.get("username", "")).strip()
        with db() as con:
            own_map(con, mid, user)
            target = con.execute("SELECT id, username FROM users WHERE username=?", (name,)).fetchone()
            if not target:
                raise HTTPException(404, "No user with this name")
            if target["id"] == user["id"]:
                raise HTTPException(422, "This is your own map")
            if con.execute("SELECT COUNT(*) FROM map_shares WHERE map_id=?", (mid,)).fetchone()[0] >= MAX_SHARES:
                raise HTTPException(409, "Shared with too many users")
            con.execute("INSERT OR IGNORE INTO map_shares VALUES (?,?)", (mid, target["id"]))
        return {"username": target["username"]}

    @api.delete("/maps/{mid}/shares/{username}")
    def unshare_map(mid: str, username: str, user: dict = Depends(current_user)):
        with db() as con:
            own_map(con, mid, user)
            con.execute("DELETE FROM map_shares WHERE map_id=? AND user_id=(SELECT id FROM users WHERE username=?)", (mid, username))
        return {"ok": True}

    @api.post("/maps/{mid}/copy")
    def copy_map(mid: str, user: dict = Depends(current_user)):
        """Your own editable copy (image, grid, tokens, walls) of a map you own or that was shared with you."""
        with db() as con:
            r = usable_map(con, mid, user)
            if con.execute("SELECT COUNT(*) FROM maps WHERE owner_id=?", (user["id"],)).fetchone()[0] >= MAX_MAPS:
                raise HTTPException(409, "Map limit reached, delete one first")
            nid = uuid.uuid4().hex[:12]
            shutil.copyfile(path(mid), path(nid))
            con.execute("INSERT INTO maps (id, owner_id, name, w, h, grid, tokens, walls, sight, created) VALUES (?,?,?,?,?,?,?,?,?,?)",
                        (nid, user["id"], f"{r['name']} (copy)"[:60], r["w"], r["h"], r["grid"], r["tokens"], r["walls"], r["sight"], time.time()))
        return {"id": nid}

    def image_response(p: Path):
        if not p.exists():
            raise HTTPException(404, "Image missing")
        return Response(p.read_bytes(), media_type="image/webp", headers={"Cache-Control": "private, max-age=3600"})

    # ---- the map on a lobby's table

    def sync_players(con, lid: str):
        """One token per member (unplaced until the player drags it onto the map); tokens of people who left are removed."""
        members = con.execute("SELECT user_id FROM lobby_members WHERE lobby_id=?", (lid,)).fetchall()
        have = {r["user_id"] for r in con.execute("SELECT user_id FROM lobby_tokens WHERE lobby_id=? AND user_id IS NOT NULL", (lid,))}
        ids = {m["user_id"] for m in members}
        for uid in ids - have:
            con.execute("INSERT INTO lobby_tokens (id, lobby_id, user_id, name, size) VALUES (?,?,?,?,1)", (uuid.uuid4().hex[:12], lid, uid, ""))
        for uid in have - ids:
            con.execute("DELETE FROM lobby_tokens WHERE lobby_id=? AND user_id=?", (lid, uid))

    def lobby_map(con, lobby):
        return con.execute("SELECT * FROM maps WHERE id=?", (lobby["map_id"],)).fetchone() if lobby["map_id"] else None

    def view_ctx(con, lobby, m, user, as_name=None):
        """Whose eyes the table is shown through. Players: their own token. DM: everything, or (as=name) one player's view."""
        is_dm = lobby["owner_id"] == user["id"]
        viewer = None
        if is_dm and as_name:
            row = con.execute("""SELECT m.user_id FROM lobby_members m JOIN users u ON u.id = m.user_id
                                 WHERE m.lobby_id=? AND u.username=?""", (lobby["id"], as_name)).fetchone()
            if not row:
                raise HTTPException(404, "No such player in this lobby")
            viewer = row["user_id"]
        elif not is_dm:
            viewer = user["id"]
        walls = json.loads(m["walls"])
        doors = json.loads(lobby["doors"] or "{}")
        ctx = {"is_dm": is_dm, "viewer": viewer, "walls": walls, "doors": doors, "los": False, "empty": False, "poly": None,
               "rev": "full:" + m["id"], "key": None}
        if viewer is None:
            return ctx
        ctx["rev"] = "open:" + m["id"]
        if not (lobby["los"] and walls):
            return ctx
        tok = con.execute("SELECT x, y FROM lobby_tokens WHERE lobby_id=? AND user_id=?", (lobby["id"], viewer)).fetchone()
        ctx["los"] = True
        if not tok or tok["x"] is None:  # not on the map yet: sees nothing
            ctx.update(empty=True, rev="empty")
            return ctx
        x, y = round(tok["x"]), round(tok["y"])
        blocking = [(w["x1"], w["y1"], w["x2"], w["y2"]) for w in walls if w["kind"] == "wall" or not doors.get(w["id"])]
        radius = round(m["sight"] * m["grid"], 1)
        key = hashlib.sha1(json.dumps([m["id"], blocking, radius, x, y]).encode()).hexdigest()
        poly = vision.cached_polygon(key, lambda: vision.polygon((x, y), blocking, m["w"], m["h"], radius))
        ctx.update(poly=poly, key=key, rev=key[:12])
        return ctx

    def tokens_for(con, lobby, user, ctx) -> list:
        sync_players(con, lobby["id"])
        rows = con.execute("""SELECT t.*, u.username, m.color AS mcolor, c.data FROM lobby_tokens t
                              LEFT JOIN users u ON u.id = t.user_id
                              LEFT JOIN lobby_members m ON m.lobby_id = t.lobby_id AND m.user_id = t.user_id
                              LEFT JOIN characters c ON c.id = m.character_id
                              WHERE t.lobby_id=? ORDER BY t.rowid""", (lobby["id"],)).fetchall()
        full = ctx["viewer"] is None
        out = []
        for r in rows:
            player = r["user_id"] is not None
            own = player and r["user_id"] == ctx["viewer"]
            if not full and not own:
                if r["hidden"] or r["x"] is None:
                    continue
                if ctx["los"] and (ctx["empty"] or not vision.inside((r["x"], r["y"]), ctx["poly"])):
                    continue
            name = r["name"]
            if player:
                try:
                    name = (json.loads(r["data"]).get("name") if r["data"] else "") or r["username"]
                except ValueError:
                    name = r["username"]
            pt = ""
            if player and r["data"]:
                try:
                    pt = portrait_tag(json.loads(r["data"]).get("portrait"))
                except ValueError:
                    pass
            out.append({"id": r["id"], "kind": "player" if player else "npc", "username": r["username"], "name": name, "portrait": pt,
                        "x": r["x"], "y": r["y"], "size": r["size"], "hidden": bool(r["hidden"]),
                        "color": (r["mcolor"] or "#999999") if player else r["color"],
                        "can_move": ctx["is_dm"] or (r["user_id"] == user["id"] and not (ctx["los"] and r["x"] is None))})
        return out

    @api.get("/lobbies/{lid}/map")
    def read_lobby_map(lid: str, as_: str = Query("", alias="as"), user: dict = Depends(current_user)):
        """Map and tokens as the viewer may see them. Walls are only sent to the DM (in the full view)."""
        with db() as con:
            lobby = lobby_for(con, lid, user)
            m = lobby_map(con, lobby)
            if not m:
                return {"map": None, "tokens": []}
            ctx = view_ctx(con, lobby, m, user, as_)
            res = {"map": meta(m), "tokens": tokens_for(con, lobby, user, ctx),
                   "vision": {"los": ctx["los"], "empty": ctx["empty"], "rev": ctx["rev"]}}
            if ctx["is_dm"]:
                res["los_on"] = bool(lobby["los"])
                if ctx["viewer"] is None:
                    res["walls"] = [{**w, "open": bool(ctx["doors"].get(w["id"]))} for w in ctx["walls"]]
            return res

    @api.get("/lobbies/{lid}/map/view")
    def lobby_map_view(lid: str, as_: str = Query("", alias="as"), user: dict = Depends(current_user)):
        """The map image for this viewer. With line of sight on, only the part their character sees is sent
        (cut out on the server, the rest is black), together with its position in the X-Crop header (x,y,w,h)."""
        with db() as con:
            lobby = lobby_for(con, lid, user)
            m = lobby_map(con, lobby)
            if not m:
                raise HTTPException(404, "No map")
            ctx = view_ctx(con, lobby, m, user, as_)
        if ctx["empty"]:
            raise HTTPException(404, "Nothing to see yet")
        if ctx["los"]:
            data, (x, y, w, h) = vision.cut(path(m["id"]), ctx["poly"], ctx["key"])
            return Response(data, media_type="image/webp", headers={"X-Crop": f"{x},{y},{w},{h}", "Cache-Control": "no-store"})
        return Response(path(m["id"]).read_bytes(), media_type="image/webp",
                        headers={"X-Crop": f"0,0,{m['w']},{m['h']}", "Cache-Control": "private, max-age=3600"})

    @api.put("/lobbies/{lid}/doors/{wid}")
    def set_door(lid: str, wid: str, body: dict = Body(...), user: dict = Depends(current_user)):
        with db() as con:
            lobby = lobby_for(con, lid, user)
            dm_only(lobby, user)
            m = need_map(con, lobby)
            if not any(w["id"] == wid and w["kind"] == "door" for w in json.loads(m["walls"])):
                raise HTTPException(404, "No such door")
            doors = json.loads(lobby["doors"] or "{}")
            if body.get("open"):
                doors[wid] = True
            else:
                doors.pop(wid, None)
            con.execute("UPDATE lobbies SET doors=? WHERE id=?", (json.dumps(doors), lid))
        hub.notify(lid, "map")
        return {"ok": True}

    @api.put("/lobbies/{lid}/sight")
    def set_sight(lid: str, body: dict = Body(...), user: dict = Depends(current_user)):
        """DM: switch line of sight on or off for the table (off: players see the whole map)."""
        with db() as con:
            dm_only(lobby_for(con, lid, user), user)
            con.execute("UPDATE lobbies SET los=? WHERE id=?", (1 if body.get("on") else 0, lid))
        hub.notify(lid, "map")
        return {"ok": True}

    @api.put("/lobbies/{lid}/map")
    def set_lobby_map(lid: str, body: dict = Body(...), user: dict = Depends(current_user)):
        """DM: show a map from the library (or null to clear the table). The map's prepared tokens are copied to the lobby."""
        with db() as con:
            dm_only(lobby_for(con, lid, user), user)
            mid = body.get("map_id")
            con.execute("DELETE FROM lobby_tokens WHERE lobby_id=?", (lid,))
            if mid:
                r = usable_map(con, str(mid), user)
                for t in json.loads(r["tokens"]):
                    con.execute("INSERT INTO lobby_tokens (id, lobby_id, name, x, y, color, size, hidden) VALUES (?,?,?,?,?,?,?,?)",
                                (uuid.uuid4().hex[:12], lid, t["name"], t["x"], t["y"], t["color"], t["size"], int(t["hidden"])))
                sync_players(con, lid)
                event(con, lid, f"The DM put the map \"{r['name']}\" on the table", user["id"])
            else:
                event(con, lid, "The DM cleared the table", user["id"])
            con.execute("UPDATE lobbies SET map_id=?, doors='{}' WHERE id=?", (str(mid) if mid else None, lid))
        hub.notify(lid, "map")
        return {"ok": True}

    def need_map(con, lobby):
        m = lobby_map(con, lobby)
        if not m:
            raise HTTPException(409, "There is no map on the table")
        return m

    @api.post("/lobbies/{lid}/tokens")
    def add_token(lid: str, body: dict = Body(...), user: dict = Depends(current_user)):
        with db() as con:
            lobby = lobby_for(con, lid, user)
            dm_only(lobby, user)
            m = need_map(con, lobby)
            if con.execute("SELECT COUNT(*) FROM lobby_tokens WHERE lobby_id=?", (lid,)).fetchone()[0] >= MAX_TOKENS:
                raise HTTPException(409, "Too many tokens")
            t = clean_token(body, m["w"], m["h"], keep_id=False)
            con.execute("INSERT INTO lobby_tokens (id, lobby_id, name, x, y, color, size, hidden) VALUES (?,?,?,?,?,?,?,?)",
                        (t["id"], lid, t["name"], t["x"], t["y"], t["color"], t["size"], int(t["hidden"])))
        hub.notify(lid, "map")
        return {"id": t["id"]}

    @api.put("/lobbies/{lid}/tokens/{tid}")
    def move_token(lid: str, tid: str, body: dict = Body(...), user: dict = Depends(current_user)):
        """Players move their own token; the DM moves any and can also rename, recolour, resize or hide NPC tokens."""
        with db() as con:
            lobby = lobby_for(con, lid, user)
            is_dm = lobby["owner_id"] == user["id"]
            m = need_map(con, lobby)
            t = con.execute("SELECT * FROM lobby_tokens WHERE id=? AND lobby_id=?", (tid, lid)).fetchone()
            if not t or (not is_dm and t["user_id"] != user["id"]):
                raise HTTPException(404, "Token not found")
            if not is_dm and t["x"] is None and lobby["los"] and json.loads(m["walls"]):
                raise HTTPException(403, "The DM puts your token on the map")  # with line of sight on you could not see where
            merged = {"name": t["name"], "x": t["x"], "y": t["y"], "color": t["color"], "size": t["size"], "hidden": t["hidden"]}
            if "x" in body or "y" in body:
                merged["x"], merged["y"] = body.get("x"), body.get("y")
            if is_dm and t["user_id"] is None:
                for k in ("name", "color", "size", "hidden"):
                    if k in body:
                        merged[k] = body[k]
            elif is_dm and "size" in body:
                merged["size"] = body["size"]
            c = clean_token(merged, m["w"], m["h"])
            con.execute("UPDATE lobby_tokens SET name=?, x=?, y=?, color=?, size=?, hidden=? WHERE id=?",
                        (c["name"] if t["user_id"] is None else t["name"], c["x"], c["y"], c["color"], c["size"], int(c["hidden"]), tid))
        hub.notify(lid, "map")
        return {"ok": True}

    @api.delete("/lobbies/{lid}/tokens/{tid}")
    def delete_token(lid: str, tid: str, user: dict = Depends(current_user)):
        with db() as con:
            dm_only(lobby_for(con, lid, user), user)
            con.execute("DELETE FROM lobby_tokens WHERE id=? AND lobby_id=? AND user_id IS NULL", (tid, lid))
        hub.notify(lid, "map")
        return {"ok": True}

    return sync_players