#!/usr/bin/env python3 """Small, dependency-free server for Guildhall, a co-op idle game.""" 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 game APP_DIR = Path(__file__).resolve().parent STATIC_DIR = APP_DIR / "static" MAX_BODY = 16 * 1024 NAME = re.compile(r"^[\w .'-]{2,20}$") TICK_SECONDS = 15 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 / "app.py", APP_DIR / "game.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 after each change.""" def __init__(self, path: Path, speed: float): self.path = path self.lock = threading.Lock() self.rng = random.Random() 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() if row: world = json.loads(row[0]) self.world = game.migrate(world, time.time() + world["config"].get("clock_offset", 0)) self.world["config"]["speed"] = speed else: self.world = game.new_world(time.time(), speed) 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,)) 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"guildhall-{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("guildhall-*.sqlite3")) for old in snapshots[:-keep] if keep > 0 else []: old.unlink() return target def tick(self) -> None: with self.lock: now = self.now() changed = game.tick(self.world, now, self.rng) # Everyone's actions run as time passes: the fires are shared, so a craft # has to meet the fire as it was, not as it is when its player next looks. if game.catch_up(self.world, now, self.rng) or changed: self.save() def player_view(world: dict, player: dict, now: float) -> dict: game.regen(world, player, now) view = {key: value for key, value in player.items() if key not in ("last_active", "hp_at", "away")} view["hp"] = int(player["hp"]) view["max_hp"] = game.max_hp(player) view["armour"] = game.armour(player) view["reduction"] = round(game.damage_reduction(player), 3) view["ward"] = game.magic_ward(player) view["ward_reduction"] = round(game.reduction(game.magic_ward(player)), 3) view["edible"] = game.edible(player) view["blessing"] = game.blessing(player) view["fight_odds"] = {a["id"]: game.fight_odds(player, a, world) for a in game.ACTIONS if "enemy" in a} view["pp"] = int(player["pp"]) view["labour_per_action"] = game.labour_per_action(player) view["levels"] = {skill: game.skill_level(player, skill) for skill in game.SKILL_IDS} view["slots"] = game.bank_slots(world, player) view["slots_used"] = game.slots_used(world, player) view["stack"] = game.stack_size(world, player) view["power"] = game.combat_power(player) me = game.fighter(player, game.blessing(player), game.potion_effect(player, "combat")) view["fighter"] = {"max_hit": me["max_hit"], "accuracy": round(me["accuracy"]), "evasion": me["evasion"], "crit": me["crit"], "interval": me["interval"], "block": me["block"], "damage": me["damage"], "role": me["role"], "abilities": me["abilities"], "style": me["style"], "mods": me["mods"], "boosts": {ab: [game.ITEMS[i]["name"] for i in player["equip"].values() if i and ab in game.ITEMS[i].get("mods", {})] for ab in me["mods"]}} view["health_potion"] = game.health_potion(player) view["perk_costs"] = {perk: game.perk_cost(player, perk) for perk in player["perks"]} view["rent_cost"] = game.rent_cost(player) view["caravan"] = game.caravan_view(world, player, now) view["titles"] = game.titles(player) view["achievements"] = game.achievements(player) view["shown"] = game.look_view(player) if player["action"]: action, fight = game.growing(player), game.current_fight(world, player) duration = fight["duration"] if fight else game.action_time(world, player, action) view["action"] = {**player["action"], "growing": action["id"], "duration": duration} if action.get("together") and not fight: view["action"]["waiting"] = True # for a group: nothing to count down run = game.run_of(world, player) if run: view["action"]["run"] = run if fight: view["action"]["fight"] = game.fight_view(fight) party = game.party_of(world, player) if party: view["action"]["with"] = [uid for uid in party["members"] if uid != player["id"]] # A fight takes as long as it takes: show the average time to win one and rest after it. view["times"] = {action["id"]: (view["fight_odds"][action["id"]]["kill_time"] + view["fight_odds"][action["id"]]["rest"]) / game.speed(world) if "enemy" in action else game.action_time(world, player, action) for action in game.ACTIONS} view["blocked"] = {action["id"]: game.can_start(world, player, action) for action in game.ACTIONS} view["drill_per_action"] = game.drill_per_action(player) view["recon_gain"] = {action["id"]: game.recon_gain(world, player, action) for action in game.ACTIONS if action.get("special") == "recon"} level = game.skill_level(player, "firekeeping") view["firekeeping"] = {"minutes": game.fire_minutes(level), "bank": game.fire_bank(level)} # Inputs as the fires leave them: a lit fire pays the log. view["fuelled"] = {action["id"]: game.fire_inputs(world, action) for action in game.ACTIONS if action.get("fire")} return view def city_view(world: dict, now: float) -> dict: city = world["city"] buildings = {} for key, building in city["buildings"].items(): working = city["construction"] and city["construction"]["building"] == key target = game.fund_target(world, key) missing = game.fund_missing(world, key) if working: status = "building" elif key in city["queue"]: status = "queued" elif target is None: status = "max" if building["level"] >= game.BUILDINGS[key]["max"] else "capped" else: status = "funding" buildings[key] = {**building, "target": target, "missing": missing, "status": status, "cap": game.level_cap(world, key)} return { **game.population_view(world), "buildings": buildings, "queue": city["queue"], "devotion": round(city.get("devotion", 0), 1), "townsfolk_labour": round(city["population"] * game.TOWNSFOLK_LABOUR, 1), "ward": game.ward(world), "treasury": int(city["treasury"]), "stockpile": city["stockpile"], "construction": city["construction"], "starving": city["starving"], "merchant_until": city.get("merchant_until", 0), "repelled": city["repelled"], "defence": game.defence(world), "day": game.calendar_day(world, now) + 1, "fires": game.fires_view(world, now), "monuments": game.monuments_view(world), } def roster(world: dict, now: float) -> list[dict]: """What everyone can see about each other: no banks, no gold.""" people = [] for player in world["players"].values(): action = game.ACTION_BY_ID[player["action"]["id"]] if player["action"] else None game.regen(world, player, now) people.append({ "xp": player["xp"], "equip": player["equip"], "joined": player["joined"], "hp": int(player["hp"]), "max_hp": game.max_hp(player), "power": game.combat_power(player), "fighter": {key: value for key, value in game.fighter(player, game.blessing(player), {}).items() if key in ("max_hit", "accuracy", "evasion", "crit", "interval", "block", "damage")}, "armour": game.armour(player), "reduction": round(game.damage_reduction(player), 3), "ward": game.magic_ward(player), "ward_reduction": round(game.reduction(game.magic_ward(player)), 3), "favour": player["favour"], "id": player["id"], "name": player["name"], "cls": player["cls"], "total": sum(game.skill_level(player, skill) for skill in game.SKILL_IDS), "levels": {skill: game.skill_level(player, skill) for skill in game.SKILL_IDS}, "doing": action["name"] if action else None, "skill": action["skill"] if action else None, # A fight others can tag into, and who's in it together. "fight": player["action"]["id"] if game.joinable(world, player) and game.offline_limit(world, player) > now else None, "party": (player["action"] or {}).get("party"), "garrison": player["garrison"], "donated": player["donated"], "look": game.look_view(player), }) return sorted(people, key=lambda p: -p["donated"]) class Handler(BaseHTTPRequestHandler): server_version = "Guildhall" 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*)gh_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_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_response(HTTPStatus.OK) self.send_header("Content-Type", mimetypes.guess_type(target.name)[0] or "application/octet-stream") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-cache") self.end_headers() self.wfile.write(body) # --- routes ---------------------------------------------------------- def do_GET(self) -> None: path = urlparse(self.path).path if path == "/api/content": self.send_json(game.content()) elif path == "/api/state": uid = self.identity() if not uid: return self.fail("Not signed in", HTTPStatus.UNAUTHORIZED) with self.store.lock: world = self.store.world now = self.store.now() game.tick(world, now, self.store.rng) player = world["players"].get(uid) summary = None game.catch_up(world, now, self.store.rng) if player: summary = game.take_summary(player, game.empty_summary()) game.regen(world, player, now) player["last_active"] = now game.watch_fight(world, player, now) # looking at it: your manual abilities hold self.store.save() self.send_json(self.state(uid, summary)) elif path.startswith("/api/"): self.fail("Not found", HTTPStatus.NOT_FOUND) else: 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 = self.store.world now = self.store.now() game.tick(world, now, self.store.rng) player = world["players"].get(uid) if path == "/api/register": error = self.register(uid, body, now) elif path.startswith("/api/dev/") and self.dev: error = self.dev_route(path, body) elif not player: return self.fail("Create your adventurer first", HTTPStatus.CONFLICT) else: game.catch_up(world, now, self.store.rng) # everyone, so the fires are as they really are summary = game.take_summary(player, game.empty_summary()) game.regen(world, player, now) player["last_active"] = now game.watch_fight(world, player, now) error = self.player_route(path, player, body, now) if error is None: game.process_player(world, player, now, self.store.rng) # a new fight is drawn up at once game.tick(world, now, self.store.rng) self.store.save() if error: return self.fail(error) return self.send_json(self.state(uid, summary)) self.store.save() if error: return self.fail(error) self.send_json(self.state(uid)) def player_route(self, path: str, player: dict, body: dict, now: float) -> str | None: world = self.store.world item = body.get("item") qty = body.get("qty") if path in ("/api/donate", "/api/sell", "/api/give"): if item != "gold" and item not in game.ITEMS: return "No such item" if qty == "all": qty = player["gold"] if item == "gold" else player["bank"].get(item, 0) if not isinstance(qty, int): return "Bad quantity" if path == "/api/action": return game.start_action(world, player, str(body.get("id")), now, once=bool(body.get("once"))) if path == "/api/join": return game.join_fight(world, player, str(body.get("id")), now) if path == "/api/press": target = body.get("target") return game.press(world, player, str(body.get("id")), now, str(target) if target else None) if path == "/api/target": return game.set_fight_target(world, player, str(body.get("id")), now) if path == "/api/learn": return game.learn_pattern(player, str(body.get("item"))) if path == "/api/pull": return game.pull_now(world, player, now) if path == "/api/run": return game.start_run(world, player, now) if path == "/api/manual": return game.set_manual(world, player, str(body.get("id")), bool(body.get("on")), now) if path == "/api/order": return game.post_order(world, player, body.get("kind"), str(item), qty, body.get("provide"), body.get("reward"), now) if path == "/api/order/provide": return game.provide_reagent(world, player, body.get("id"), str(item)) if path == "/api/order/cancel": return game.cancel_order(world, player, body.get("id"), now) if path == "/api/order/remove": return game.remove_orders(world, player, body.get("id")) if path == "/api/order/work": return game.work_order(world, player, body.get("id"), now) if path == "/api/order/drop": return game.claim_order(world, player, body.get("id"), claim=False) if path == "/api/order/deliver": return game.deliver_order(world, player, body.get("id"), now) if path == "/api/seen": event_id = body.get("id") if not isinstance(event_id, int): return "Bad event" player["seen_threat"] = max(player.get("seen_threat", 0), event_id) return None if path == "/api/stop": game.drop_out(world, player, now) return None if path == "/api/donate": return game.donate(world, player, item, qty, now) if path == "/api/give": receiver = world["players"].get(str(body.get("to"))) if not receiver: return "No such adventurer" if item == "gold": return "Gold goes to the treasury, not between players" return game.give(world, player, receiver, item, qty, now) if path == "/api/sell": return "Gold can't be sold" if item == "gold" else game.sell(world, player, item, qty, now) if path == "/api/equip": return self.equip(player, item, body.get("slot")) if path == "/api/fund": qty = body.get("qty") return game.fund(world, player, str(body.get("building")), str(body.get("item")), qty if qty == "all" else (qty if isinstance(qty, int) else -1), str(body.get("source")), now) if path == "/api/install": return game.install(world, player, str(body.get("fitting")), str(body.get("source")), now) if path == "/api/garrison": raid = game.upcoming_raid(world) if body.get("join") and not raid: return "There's no raid to stand guard for" player["garrison"] = bool(body.get("join")) return None if path == "/api/board/delete": message = next((m for m in world["board"] if m.get("id") == body.get("id")), None) if not message: return "That message is gone" if message["who"] != player["id"]: return "You can only delete your own messages" world["board"].remove(message) return None if path == "/api/aid": return game.aid_problem(world, player, str(body.get("kind")), str(body.get("item")), str(body.get("source")), now) if path == "/api/hire": return game.hire_mercenaries(world, player, now) if path == "/api/wages": return game.pay_wages(world, player, qty if qty == "all" or isinstance(qty, int) else -1, str(body.get("source")), now) if path == "/api/rent": return game.rent_slot(player) if path == "/api/buy": skill = body.get("skill") return game.buy_ware(world, player, str(body.get("ware")), str(skill) if skill else None, now) if path == "/api/look": look = body.get("look") return game.set_look(player, look) if isinstance(look, dict) else "Bad request" if path == "/api/home": return game.upgrade_home(player) if path == "/api/monument": return game.give_monument(world, player, str(body.get("id")), qty if qty == "all" or isinstance(qty, int) else -1, now) if path == "/api/statue": return game.carve_statue(world, player, str(body.get("of")), now) if path == "/api/potion": return game.use_potion(player, str(body.get("item"))) if path == "/api/prayers": ids = body.get("ids") return game.set_prayers(player, ids) if isinstance(ids, list) else "Bad request" if path == "/api/eat": return game.set_eat_below(player, body.get("below")) if path == "/api/food": food = body.get("item") if food is not None and "heal" not in game.ITEMS.get(str(food), {}): return "That isn't food" player["food"] = food return None if path == "/api/favour": return game.spend_favour(world, player, str(body.get("kind")), now) if path == "/api/class": cls = body.get("cls") if not player.get("repick"): return "You've already chosen your trade" if cls not in game.CLASSES: return "Pick a class" player["cls"], player["repick"] = cls, False game.log(world, now, "join", f"{player['name']} took up the {game.CLASSES[cls]['name'].lower()}'s trade", player["id"]) return None if path == "/api/plot": return game.plant_second_plot(world, player, body.get("id") or None) if path == "/api/perk": return game.buy_perk(player, str(body.get("perk"))) if path == "/api/pin": return game.toggle_pin(player, str(body.get("skill"))) if path == "/api/track": return game.toggle_track(player, str(body.get("id"))) if path == "/api/settings": if body.get("overflow") not in ("city", "stop"): return "Bad setting" player["overflow"] = body["overflow"] return None if path == "/api/board": text = str(body.get("text", "")).strip()[:500] if not text: return "Say something" world["next_message_id"] = world.get("next_message_id", 1) + 1 world["board"].append({"id": world["next_message_id"] - 1, "at": now, "who": player["id"], "text": text}) del world["board"][:-game.MAX_LOG] return None return "Not found" def equip(self, player: dict, item: str | None, slot: str | None) -> str | None: # A smaller strongbox or bag can't go on while the bank holds more than it would. world = self.store.world over, bank, equip = game.overfull(world, player), dict(player["bank"]), dict(player["equip"]) error = self._equip(player, item, slot) if not error and game.overfull(world, player) > max(0, over): player["bank"], player["equip"] = bank, equip return "Bank too full for that" return error def _equip(self, player: dict, item: str | None, slot: str | None) -> str | None: if item is None: if slot not in player["equip"]: return "No such slot" current = player["equip"][slot] if current: if game.bank_room(self.store.world, player, current) < 1: return "No room in the bank" player["bank"][current] = player["bank"].get(current, 0) + 1 player["equip"][slot] = None return None spec = game.ITEMS.get(item) if not spec or "slot" not in spec: return "That can't be equipped" if player["bank"].get(item, 0) < 1: return "You don't have one" problem = game.equip_problem(player, item) if problem: return problem shield = player["equip"].get("shield") if spec.get("two_handed") and shield: # Both hands on the weapon: the shield goes back to the bank. if game.bank_room(self.store.world, player, shield) < 1: return "No room in the bank for your shield" player["bank"][shield] = player["bank"].get(shield, 0) + 1 player["equip"]["shield"] = None game.bank_take(player, item, 1) current = player["equip"][spec["slot"]] if current: if game.bank_room(self.store.world, player, current) < 1: player["bank"][item] = player["bank"].get(item, 0) + 1 return "No room in the bank for what you're wearing" player["bank"][current] = player["bank"].get(current, 0) + 1 player["equip"][spec["slot"]] = item return None def register(self, uid: str, body: dict, now: float) -> str | None: world = self.store.world if uid in world["players"]: return "You already have an adventurer" name = str(body.get("name", "")).strip() cls = body.get("cls") if not NAME.match(name): return "Pick a name of 2 to 20 letters" if cls not in game.CLASSES: return "Pick a class" world["players"][uid] = game.new_player(uid, name, cls, now) game.log(world, now, "join", f"{name} the {game.CLASSES[cls]['name'].lower()} arrived in the city", uid) return None def dev_route(self, path: str, body: dict) -> str | None: world = self.store.world if path == "/api/dev/advance": hours = float(body.get("hours", 1)) world["config"]["clock_offset"] += hours * game.HOUR / game.speed(world) game.tick(world, self.store.now(), self.store.rng) return None if path == "/api/dev/happen": track = body.get("track") fire = {"minor": game._minor_threat, "hardship": game._hardship, "fortune": game._fortune}.get(track) if not fire: return "No such track" fire(world, self.store.now(), self.store.rng, body.get("kind")) return None if path == "/api/dev/event": world["next_event_at"] = self.store.now() game.tick(world, self.store.now(), self.store.rng) return None return "Not found" def state(self, uid: str, summary: dict | None = None) -> dict: world = self.store.world now = self.store.now() player = world["players"].get(uid) upcoming = game.upcoming_event(world) unseen = game.unseen_threat(world, player) if player else None # Newest first by when it happened: raids are announced long before they land. done = sorted((e for e in world["events"] if e["status"] == "done"), key=lambda e: e["resolved"], reverse=True) history = [ {"id": e["id"], "kind": e["kind"], "track": e.get("track", "major"), "at": e["resolved"], "report": e.get("report")} for e in done ][:15] return { "now": now, "dev": self.dev, "uid": uid, "version": VERSION, "me": player_view(world, player, now) if player else None, "summary": summary, "city": city_view(world, now), "upcoming": game.event_view(world, upcoming, now) if upcoming else None, "favours": game.favours(world, now), "happenings": game.happenings_view(world, now), "history": history, "orders": world["orders"], "threat_report": unseen and {"id": unseen["id"], "kind": unseen["kind"], "at": unseen["resolved"], "report": unseen["report"]}, "players": roster(world, now), "ledger": world["ledger"][-80:], "board": world["board"][-80:], } def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--host", default="127.0.0.1") parser.add_argument("--port", type=int, default=8098) parser.add_argument("--db", type=Path, default=Path("guildhall.sqlite3")) parser.add_argument("--speed", type=float, default=1.0, help="game time multiplier (dev worlds)") parser.add_argument("--dev", action="store_true", help="trust a gh_user cookie and enable time travel") 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.speed) 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("Guildhall listening on http://%s:%d%s", args.host, args.port, " (dev)" if args.dev else "") server.serve_forever() if __name__ == "__main__": main()