wog/app.py

#!/usr/bin/env python3
"""Small, dependency-free server for World of Guildhall."""

from __future__ import annotations

import argparse
import hashlib
import json
import logging
import mimetypes
import random
import re
import sqlite3
import threading
import time
from contextlib import closing
from datetime import date
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import unquote, urlparse

import combat
import content
import game

APP_DIR = Path(__file__).resolve().parent
STATIC_DIR = APP_DIR / "static"
MAX_BODY = 16 * 1024
NAME = re.compile(r"^[A-Za-z][A-Za-z'-]{1,15}$")
TICK_SECONDS = 0.25
SAVE_SECONDS = 5


def build_version() -> str:
    """Changes whenever the code or client does; open tabs reload when it does."""
    digest = hashlib.sha256()
    for path in sorted([*APP_DIR.glob("*.py"), *STATIC_DIR.rglob("*")]):
        if path.is_file():
            digest.update(path.name.encode())
            digest.update(path.read_bytes())
    return digest.hexdigest()[:12]


VERSION = build_version()

mimetypes.add_type("application/manifest+json", ".webmanifest")


class Store:
    """The world lives in memory and is written whole to SQLite every few seconds."""

    def __init__(self, path: Path, dev: bool):
        self.path = path
        self.lock = threading.Lock()
        self.rng = random.Random()
        self.saved = 0.0
        path.parent.mkdir(parents=True, exist_ok=True)
        with closing(sqlite3.connect(path)) as db, db:
            db.execute("CREATE TABLE IF NOT EXISTS world (id INTEGER PRIMARY KEY CHECK (id = 1), data TEXT NOT NULL)")
            row = db.execute("SELECT data FROM world WHERE id = 1").fetchone()
        self.world = game.migrate(json.loads(row[0])) if row else game.new_world(time.time())
        self.world["config"]["dev"] = dev
        self.save()

    def now(self) -> float:
        return time.time() + self.world["config"]["clock_offset"]

    def save(self) -> None:
        data = json.dumps(self.world, separators=(",", ":"))
        with closing(sqlite3.connect(self.path)) as db, db:
            db.execute("INSERT INTO world (id, data) VALUES (1, ?) ON CONFLICT (id) DO UPDATE SET data = excluded.data",
                       (data,))
        self.saved = time.time()

    def backup(self, destination: Path, keep: int) -> Path:
        """Write today's consistent snapshot and prune all but the newest `keep`."""
        destination.mkdir(parents=True, exist_ok=True)
        target = destination / f"wog-{date.today().isoformat()}.sqlite3"
        partial = target.with_suffix(".partial")
        with closing(sqlite3.connect(self.path)) as source, closing(sqlite3.connect(partial)) as copy:
            source.backup(copy)
        partial.replace(target)
        snapshots = sorted(destination.glob("wog-*.sqlite3"))
        for old in snapshots[:-keep] if keep > 0 else []:
            old.unlink()
        return target

    def tick(self) -> None:
        with self.lock:
            game.process(self.world, self.now(), self.rng)
            if time.time() - self.saved > SAVE_SECONDS:
                self.save()


def static_content() -> dict:
    referenced = set(content.ITEMS)
    for quest in content.QUESTS.values():
        referenced.update(quest.get("choice", []))
    return {
        "classes": content.CLASSES, "specs": content.SPECS, "abilities": content.ABILITIES, "unlock": content.UNLOCK,
        "places": content.PLACES, "zones": content.ZONES, "roads": content.ROADS, "flight_cost": content.FLIGHT_COST,
        "quests": content.QUESTS, "dungeons": content.DUNGEONS,
        "mobs": {k: {"name": m["name"], "lv": m["lv"], "rare": m.get("rare", False), "boss": m.get("boss", False),
                     "elite": m.get("elite", False)} for k, m in content.MOBS.items()},
        "enemy_abilities": content.ENEMY_ABILITIES, "slots": content.SLOTS, "slot_names": content.SLOT_NAMES,
        "xp": content.XP_TO_NEXT, "max_level": content.MAX_LEVEL,
        "items": {i: game.item(i) for i in sorted(referenced)},
        "auto_cd": combat.AUTO_CD, "watch": combat.WATCH, "gcd": combat.GCD,
    }


CONTENT = json.dumps(static_content(), separators=(",", ":")).encode()


def player_view(world: dict, player: dict, now: float) -> dict:
    st = game.stats(player)
    hp, mana = game.hp_now(player, now)
    view = {key: player[key] for key in ("id", "name", "cls", "level", "xp", "copper", "place", "zones", "activity",
                                          "quests", "done", "manual", "group", "dead_until", "next_pull")}
    view["hp"], view["mana"] = round(hp), round(mana, 3)
    view["stats"] = st
    view["xp_next"] = content.XP_TO_NEXT[player["level"]]
    view["known"] = game.known(player)
    view["equip"] = {slot: game.item(i) if i else None for slot, i in player["equip"].items()}
    bags = []
    for item_id, qty in sorted(player["bags"].items()):
        spec = game.item(item_id)
        if spec:
            bags.append({**spec, "qty": qty, "fit": game.can_wear(player, spec) if "slot" in spec else None})
    view["bags"] = bags
    view["available"] = game.available_quests(player)
    view["log"] = player["log"][-40:]
    return view


def group_view(world: dict, group: dict | None, now: float) -> dict | None:
    if not group:
        return None
    members = []
    for uid in group["members"]:
        member = world["players"][uid]
        hp, mana = game.hp_now(member, now)
        members.append({"id": uid, "name": member["name"], "cls": member["cls"], "level": member["level"],
                        "place": member["place"], "hp": round(hp), "max_hp": game.stats(member)["max_hp"],
                        "mana": round(mana, 2), "inside": bool(member["activity"]
                                                                and member["activity"]["kind"] == "dungeon"),
                        "busy": game._busy(member)})
    dungeon = {k: v for k, v in group["dungeon"].items() if k != "last_fight"} if group["dungeon"] else None
    return {"id": group["id"], "leader": group["leader"], "dungeon": dungeon, "members": members}


def roster(world: dict, now: float) -> list[dict]:
    people = []
    for player in world["players"].values():
        activity = player["activity"]
        people.append({"id": player["id"], "name": player["name"], "cls": player["cls"], "level": player["level"],
                       "place": player["place"], "group": player["group"],
                       "doing": activity["kind"] if activity else None,
                       "hunt": activity.get("hunt") if activity else None,
                       "online": now - player["last_active"] < 120})
    return sorted(people, key=lambda p: (-p["level"], p["name"]))


class Handler(BaseHTTPRequestHandler):
    server_version = "WoG"
    store: Store
    dev = False
    user_header = "Remote-User"

    def log_message(self, fmt: str, *args) -> None:
        logging.debug("%s - %s", self.address_string(), fmt % args)

    # --- plumbing --------------------------------------------------------

    def identity(self) -> str | None:
        if self.dev:
            cookie = self.headers.get("Cookie", "")
            match = re.search(r"(?:^|;\s*)wog_user=([^;]+)", cookie)
            return unquote(match.group(1)) if match else "fred"
        user = self.headers.get(self.user_header, "").strip()
        return user or None

    def send_bytes(self, body: bytes, kind: str = "application/json", cache: str = "no-store") -> None:
        self.send_response(HTTPStatus.OK)
        self.send_header("Content-Type", kind)
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", cache)
        self.end_headers()
        self.wfile.write(body)

    def send_json(self, payload: object, status: HTTPStatus = HTTPStatus.OK) -> None:
        body = json.dumps(payload, separators=(",", ":")).encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(body)

    def fail(self, message: str, status: HTTPStatus = HTTPStatus.BAD_REQUEST) -> None:
        self.send_json({"error": message}, status)

    def read_body(self) -> dict | None:
        length = int(self.headers.get("Content-Length") or 0)
        if length > MAX_BODY:
            return None
        try:
            data = json.loads(self.rfile.read(length) or b"{}")
        except json.JSONDecodeError:
            return None
        return data if isinstance(data, dict) else None

    def send_static(self, path: str) -> None:
        name = "index.html" if path in ("", "/") else path.lstrip("/").removeprefix("static/")
        target = (STATIC_DIR / name).resolve()
        if STATIC_DIR.resolve() not in target.parents or not target.is_file():
            target = STATIC_DIR / "index.html"
        body = target.read_bytes()
        if target.name == "index.html":
            # Versioned URLs, so no cache (a home-screen app's included) can serve an old client.
            for asset in (b"/static/app.js", b"/static/styles.css"):
                body = body.replace(asset + b'"', asset + b"?v=" + VERSION.encode() + b'"')
        self.send_bytes(body, mimetypes.guess_type(target.name)[0] or "application/octet-stream", "no-cache")

    # --- routes ----------------------------------------------------------

    def do_GET(self) -> None:
        path = urlparse(self.path).path
        if path == "/api/content":
            return self.send_bytes(CONTENT, cache="no-cache")
        if path == "/api/state":
            uid = self.identity()
            if not uid:
                return self.fail("Not signed in", HTTPStatus.UNAUTHORIZED)
            with self.store.lock:
                world, now = self.store.world, self.store.now()
                player = world["players"].get(uid)
                if player:
                    player["last_active"] = player["watch"] = now
                game.process(world, now, self.store.rng)
                return self.send_json(self.state(uid))
        if path.startswith("/api/"):
            return self.fail("Not found", HTTPStatus.NOT_FOUND)
        self.send_static(path)

    def do_POST(self) -> None:
        path = urlparse(self.path).path
        uid = self.identity()
        if not uid:
            return self.fail("Not signed in", HTTPStatus.UNAUTHORIZED)
        body = self.read_body()
        if body is None:
            return self.fail("Bad request body")
        with self.store.lock:
            world, now = self.store.world, self.store.now()
            game.process(world, now, self.store.rng)
            player = world["players"].get(uid)
            if path == "/api/register":
                error = self.register(uid, body, now)
            elif not player:
                return self.fail("Create your character first", HTTPStatus.CONFLICT)
            elif path.startswith("/api/dev/"):
                error = self.dev_route(path, player, body, now) if self.dev else "Not found"
            else:
                player["last_active"] = player["watch"] = now
                error = self.player_route(path, player, body, now)
            if error:
                return self.fail(error)
            game.process(world, now, self.store.rng)
            if path not in ("/api/press", "/api/target"):
                self.store.save()
            self.send_json(self.state(uid))

    def player_route(self, path: str, player: dict, body: dict, now: float) -> str | None:
        world = self.store.world
        arg = body.get("id")
        arg = str(arg) if arg is not None else ""
        if path == "/api/press":
            fight = game.current_fight(world, player)
            if not fight:
                return "You're not fighting"
            target = body.get("target")
            return combat.press(fight, player["id"], arg, now, str(target) if target else None)
        if path == "/api/target":
            fight = game.current_fight(world, player)
            return combat.set_target(fight, player["id"], arg) if fight else "You're not fighting"
        if path == "/api/manual":
            return game.set_manual(world, player, arg, bool(body.get("on")))
        if path == "/api/hunt":
            return game.start_hunt(world, player, arg, now)
        if path == "/api/stop":
            return game.stop(world, player, now)
        if path == "/api/journey":
            return game.journey(world, player, arg, now)
        if path == "/api/fly":
            return game.fly(world, player, arg, now)
        if path == "/api/equip":
            return game.equip(player, arg)
        if path == "/api/unequip":
            return game.unequip(player, arg)
        if path == "/api/sell":
            qty = body.get("qty", 1)
            return game.sell(player, arg, qty if qty == "all" else (qty if isinstance(qty, int) else -1))
        if path == "/api/sell_junk":
            return game.sell_junk(player)
        if path == "/api/accept":
            return game.accept(player, arg)
        if path == "/api/abandon":
            return game.abandon(player, arg)
        if path == "/api/turn_in":
            return game.turn_in(world, player, arg, body.get("choice"), now)
        if path == "/api/group/create":
            return game.create_group(world, player)
        if path == "/api/group/join":
            return game.join_group(world, player, arg)
        if path == "/api/group/leave":
            return game.leave_group(world, player, now)
        if path == "/api/dungeon/enter":
            return game.enter_dungeon(world, player, now)
        if path == "/api/dungeon/pull":
            return game.pull_now(world, player, now)
        if path == "/api/dungeon/leave":
            return game.leave_dungeon(world, player, now)
        if path == "/api/say":
            return game.say(world, player, str(body.get("text", "")), now)
        return "Not found"

    def register(self, uid: str, body: dict, now: float) -> str | None:
        world = self.store.world
        if uid in world["players"]:
            return "You already have a character"
        name = str(body.get("name", "")).strip().capitalize()
        cls = body.get("cls")
        if not NAME.match(name):
            return "Names are 2 to 16 letters"
        if any(p["name"].lower() == name.lower() for p in world["players"].values()):
            return "That name is taken"
        if cls not in content.CLASSES:
            return "Pick a class"
        world["players"][uid] = game.new_player(uid, name, cls, now)
        game.world_log(world, now, f"{name} the {content.CLASSES[cls]['name']} arrived in Northshire", "join", uid)
        return None

    def dev_route(self, path: str, player: dict, body: dict, now: float) -> str | None:
        world = self.store.world
        if path == "/api/dev/level":
            return game.dev_level(world, player, int(body.get("level", 1)), now)
        if path == "/api/dev/advance":
            world["config"]["clock_offset"] += float(body.get("minutes", 10)) * game.MINUTE
            return None
        if path == "/api/dev/teleport":
            if str(body.get("id")) not in content.PLACES:
                return "No such place"
            game.stop(world, player, now)
            player["place"] = str(body["id"])
            if game.zone_of(player["place"]) not in player["zones"]:
                player["zones"].append(game.zone_of(player["place"]))
            return None
        return "Not found"

    def state(self, uid: str) -> dict:
        world = self.store.world
        now = self.store.now()
        player = world["players"].get(uid)
        fight = game.shown_fight(world, player, now) if player else None
        group = world["groups"].get(player["group"] or "") if player else None
        return {
            "now": now, "dev": self.dev, "uid": uid, "version": VERSION,
            "me": player_view(world, player, now) if player else None,
            "fight": combat.view(fight) if fight else None,
            "group": group_view(world, group, now),
            "groups": [{"id": g["id"], "leader": world["players"][g["leader"]]["name"], "size": len(g["members"]),
                        "inside": bool(g["dungeon"])} for g in world["groups"].values()],
            "players": roster(world, now),
            "log": world["log"][-30:],
            "chat": world["chat"][-60:],
        }


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=8100)
    parser.add_argument("--db", type=Path, default=Path("wog.sqlite3"))
    parser.add_argument("--dev", action="store_true", help="trust a wog_user cookie and enable cheats")
    parser.add_argument("--user-header", default="Remote-User")
    parser.add_argument("--backup-to", type=Path, help="write a snapshot of the database to this directory and exit")
    parser.add_argument("--keep", type=int, default=30, help="snapshots to keep with --backup-to")
    args = parser.parse_args()
    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")

    store = Store(args.db, args.dev)
    if args.backup_to:
        logging.info("Wrote %s", store.backup(args.backup_to, args.keep))
        return
    Handler.store = store
    Handler.dev = args.dev
    Handler.user_header = args.user_header

    def ticker() -> None:
        while True:
            time.sleep(TICK_SECONDS)
            try:
                store.tick()
            except Exception:
                logging.exception("tick failed")

    threading.Thread(target=ticker, daemon=True).start()
    server = ThreadingHTTPServer((args.host, args.port), Handler)
    logging.info("World of Guildhall listening on http://%s:%d%s", args.host, args.port, " (dev)" if args.dev else "")
    server.serve_forever()


if __name__ == "__main__":
    main()