"""The Button - a single-file, dependency-free game server (Python 3.9+)."""

import csv
import hashlib
import hmac
import json
import os
import re
import secrets
import sqlite3
import threading
import time
from contextlib import contextmanager
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

ROOT = Path(__file__).resolve().parent
STATIC_DIR = ROOT / "static"
DATA_FILE = Path(os.environ.get("BUTTON_DATA_FILE", ROOT / "data.json"))
DATABASE_FILE = Path(os.environ.get("BUTTON_DATABASE_FILE", DATA_FILE.with_suffix(".sqlite3")))
MAILING_LIST_FILE = Path(os.environ.get("BUTTON_MAILING_LIST_FILE", ROOT / "mailing_list.csv"))
MAILING_LIST_FIELDS = ["subscribed_at", "player", "first_name", "last_name", "email", "phone"]
EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
PHONE_RE = re.compile(r"^\+?[0-9 ()\-.]{7,25}$")
HOST = os.environ.get("BUTTON_HOST", "127.0.0.1")
PORT = int(os.environ.get("BUTTON_PORT", "8000"))
DURATION_MS = int(float(os.environ.get("BUTTON_DURATION_SECONDS", 24 * 60 * 60)) * 1000)
MAX_BODY_BYTES = 1024
MAX_NAME_LEN = 24
MIN_PASSWORD_LEN = 8
MAX_PASSWORD_LEN = 128
PBKDF2_ITERATIONS = 600_000
MAX_SESSIONS_PER_PLAYER = 10
LOGIN_MAX_FAILURES = 5
LOGIN_LOCKOUT_MS = 15 * 60 * 1000
MAX_CHAT_LENGTH = 200
CHAT_HISTORY_LIMIT = 100
CHAT_COOLDOWN_MS = 2000
PRESENCE_TTL_SECONDS = 45
MAX_PRESENCE_TABS_PER_SESSION = 10

# Ordered from a freshly reset countdown to an almost-expired one.
# "share" is each zone's relative slice of the timer, so high-value zones are brief.
ZONES = [
    {"name": "green", "points": 0, "share": 32},
    {"name": "blue", "points": 2, "share": 16},
    {"name": "yellow", "points": 4, "share": 8},
    {"name": "red", "points": 8, "share": 4},
    {"name": "black", "points": 16, "share": 2},
    {"name": "white", "points": 32, "share": 1},
]
_total_share = sum(z["share"] for z in ZONES)
_upper = DURATION_MS
for _z in ZONES:
    _z["remaining_from_ms"] = _upper
    _upper = round(_upper - DURATION_MS * _z["share"] / _total_share)
    _z["remaining_to_ms"] = max(_upper, 0)
ZONES[-1]["remaining_to_ms"] = 0

STATIC_FILES = {
    "/": ("index.html", "text/html; charset=utf-8"),
    "/index.html": ("index.html", "text/html; charset=utf-8"),
    "/app.js": ("app.js", "text/javascript; charset=utf-8"),
    "/style.css": ("style.css", "text/css; charset=utf-8"),
    "/favicon.ico": ("favicon.ico", "image/x-icon"),
    "/poster.png": ("poster.png", "image/png"),
}

lock = threading.Lock()


def now_ms():
    return int(time.time() * 1000)


def hash_token(token):
    return hashlib.sha256(token.encode("utf-8")).hexdigest()


def hash_password(password, salt=None, iterations=PBKDF2_ITERATIONS):
    salt = salt or secrets.token_hex(16)
    digest = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), bytes.fromhex(salt), iterations).hex()
    return f"pbkdf2_sha256${iterations}${salt}${digest}"


def verify_password(password, stored):
    _, iterations, salt, _ = stored.split("$")
    return hmac.compare_digest(hash_password(password, salt, int(iterations)), stored)


# Verified against when the name doesn't exist so response timing doesn't reveal valid names.
DUMMY_PASSWORD_HASH = hash_password(secrets.token_hex(16))


def zone_for_remaining(remaining_ms):
    for zone in ZONES:
        if remaining_ms > zone["remaining_to_ms"]:
            return zone
    return ZONES[-1]


def new_game():
    t = now_ms()
    return {
        "duration_ms": DURATION_MS,
        "started_at": t,
        "deadline": t + DURATION_MS,
        "ended": False,
        "ended_at": None,
        "players": {},  # id -> {name, password_hash, token_hashes, created_at}
        "presses": [],  # {player_id, at, remaining_ms, zone, points}
    }


@contextmanager
def database():
    """Use a separate connection per operation, with atomic writes and FK checks."""
    connection = sqlite3.connect(DATABASE_FILE)
    connection.row_factory = sqlite3.Row
    try:
        connection.execute("PRAGMA foreign_keys = ON")
        with connection:
            yield connection
    finally:
        connection.close()


def write_state(connection, state):
    """Persist a complete snapshot inside the caller's transaction."""
    connection.execute("DELETE FROM presses")
    connection.execute("DELETE FROM sessions")
    connection.execute("DELETE FROM players")
    connection.executemany(
        "INSERT INTO players (id, name, password_hash, created_at, subscribed) VALUES (?, ?, ?, ?, ?)",
        [(pid, p["name"], p.get("password_hash"), p["created_at"], p.get("subscribed", False))
         for pid, p in state["players"].items()],
    )
    connection.executemany(
        "INSERT INTO sessions (player_id, position, token_hash) VALUES (?, ?, ?)",
        [(pid, position, token_hash) for pid, p in state["players"].items()
         for position, token_hash in enumerate(p["token_hashes"])],
    )
    connection.executemany(
        "INSERT INTO presses (position, player_id, at, remaining_ms, zone, points) VALUES (?, ?, ?, ?, ?, ?)",
        [(position, p["player_id"], p["at"], p["remaining_ms"], p["zone"], p["points"])
         for position, p in enumerate(state["presses"])],
    )
    connection.execute(
        """INSERT INTO game (id, duration_ms, started_at, deadline, ended, ended_at)
           VALUES (1, ?, ?, ?, ?, ?)
           ON CONFLICT(id) DO UPDATE SET duration_ms = excluded.duration_ms,
               started_at = excluded.started_at, deadline = excluded.deadline,
               ended = excluded.ended, ended_at = excluded.ended_at""",
        (state["duration_ms"], state["started_at"], state["deadline"], state["ended"], state["ended_at"]),
    )


def load_state():
    if DATABASE_FILE.resolve() == DATA_FILE.resolve():
        raise ValueError("BUTTON_DATABASE_FILE must differ from the legacy BUTTON_DATA_FILE")
    DATABASE_FILE.parent.mkdir(parents=True, exist_ok=True)
    with database() as connection:
        connection.executescript("""
            CREATE TABLE IF NOT EXISTS game (
                id INTEGER PRIMARY KEY CHECK (id = 1),
                duration_ms INTEGER NOT NULL,
                started_at INTEGER NOT NULL,
                deadline INTEGER NOT NULL,
                ended INTEGER NOT NULL CHECK (ended IN (0, 1)),
                ended_at INTEGER
            );
            CREATE TABLE IF NOT EXISTS players (
                id TEXT PRIMARY KEY,
                name TEXT NOT NULL,
                password_hash TEXT,
                created_at INTEGER NOT NULL,
                subscribed INTEGER NOT NULL CHECK (subscribed IN (0, 1))
            );
            CREATE TABLE IF NOT EXISTS sessions (
                player_id TEXT NOT NULL REFERENCES players(id),
                position INTEGER NOT NULL,
                token_hash TEXT NOT NULL UNIQUE,
                PRIMARY KEY (player_id, position)
            );
            CREATE TABLE IF NOT EXISTS presses (
                position INTEGER PRIMARY KEY,
                player_id TEXT NOT NULL REFERENCES players(id),
                at INTEGER NOT NULL,
                remaining_ms INTEGER NOT NULL,
                zone TEXT NOT NULL,
                points INTEGER NOT NULL
            );
            CREATE TABLE IF NOT EXISTS chat_messages (
                id INTEGER PRIMARY KEY,
                player_id TEXT NOT NULL REFERENCES players(id) DEFERRABLE INITIALLY DEFERRED,
                name TEXT NOT NULL,
                at INTEGER NOT NULL,
                message TEXT NOT NULL CHECK (length(message) BETWEEN 1 AND 200)
            );
            CREATE INDEX IF NOT EXISTS chat_player_messages ON chat_messages (player_id, id);
        """)
        columns = {row["name"] for row in connection.execute("PRAGMA table_info(chat_messages)")}
        if "kind" not in columns:
            connection.execute("ALTER TABLE chat_messages ADD COLUMN kind TEXT NOT NULL DEFAULT 'message'")
        # Serialize initialization. A committed game row is the migration marker,
        # not merely the existence of a possibly empty or interrupted database.
        connection.execute("BEGIN IMMEDIATE")
        game = connection.execute("SELECT * FROM game WHERE id = 1").fetchone()
        if game is None:
            if DATA_FILE.exists():
                with DATA_FILE.open("r", encoding="utf-8") as f:
                    loaded = json.load(f)
                for player in loaded["players"].values():
                    if "token_hash" in player:  # migrate pre-login data as well
                        player["token_hashes"] = [player.pop("token_hash")]
            else:
                loaded = new_game()
            # Any read/validation/SQL failure aborts startup without consuming the
            # JSON source or committing a partial game. The next start can retry.
            write_state(connection, loaded)
            return loaded

        loaded = dict(game)
        del loaded["id"]
        loaded["ended"] = bool(loaded["ended"])
        loaded["players"] = {}
        for row in connection.execute("SELECT * FROM players ORDER BY rowid"):
            player = dict(row)
            pid = player.pop("id")
            player["subscribed"] = bool(player["subscribed"])
            player["token_hashes"] = []
            loaded["players"][pid] = player
        for row in connection.execute("SELECT * FROM sessions ORDER BY player_id, position"):
            loaded["players"][row["player_id"]]["token_hashes"].append(row["token_hash"])
        loaded["presses"] = []
        for row in connection.execute("SELECT * FROM presses ORDER BY position"):
            press = dict(row)
            del press["position"]
            loaded["presses"].append(press)
        return loaded


def save_state(state):
    with database() as connection:
        write_state(connection, state)


state = load_state()
token_index = {th: pid for pid, p in state["players"].items() for th in p["token_hashes"]}
failed_logins = {}  # lowercase name -> (failure count, first failure ms)
presence = {}  # (token hash, tab id) -> (monotonic heartbeat seconds, viewing chat)


def prune_presence():
    """Must be called with the lock held. Presence is never persisted."""
    cutoff = time.monotonic() - PRESENCE_TTL_SECONDS
    for key, (seen, _) in list(presence.items()):
        if seen <= cutoff or key[0] not in token_index:
            del presence[key]


def public_presence():
    """Public aggregates and opted-in chat names only; no session identifiers."""
    prune_presence()
    online = {token_index[th] for th, _ in presence}
    chatting = {token_index[th] for (th, _), (_, in_chat) in presence.items() if in_chat}
    return {
        "online_count": len(online),
        "chat_players": sorted((state["players"][pid]["name"] for pid in chatting), key=str.lower),
    }


def chat_messages():
    with database() as connection:
        rows = connection.execute(
            "SELECT id, name, at, message, kind FROM chat_messages ORDER BY id DESC LIMIT ?",
            (CHAT_HISTORY_LIMIT,),
        ).fetchall()
    return [dict(row) for row in reversed(rows)]


def find_player_by_name(name):
    for pid, p in state["players"].items():
        if p["name"].lower() == name.lower():
            return pid
    return None


def issue_token(pid):
    """Must be called with the lock held."""
    token = secrets.token_urlsafe(32)
    th = hash_token(token)
    hashes = state["players"][pid]["token_hashes"]
    hashes.append(th)
    token_index[th] = pid
    while len(hashes) > MAX_SESSIONS_PER_PLAYER:
        token_index.pop(hashes.pop(0), None)
    save_state(state)
    return token


def validate_credentials(data):
    """Returns (name, password, error)."""
    name, password = data.get("name"), data.get("password")
    if not isinstance(name, str) or not (1 <= len(name.strip()) <= MAX_NAME_LEN):
        return None, None, f"Name must be 1-{MAX_NAME_LEN} characters"
    name = " ".join(name.split())
    if not name.isprintable():
        return None, None, "Name contains invalid characters"
    if not isinstance(password, str) or not (MIN_PASSWORD_LEN <= len(password) <= MAX_PASSWORD_LEN):
        return None, None, f"Password must be {MIN_PASSWORD_LEN}-{MAX_PASSWORD_LEN} characters"
    return name, password, None


def check_ended():
    """Must be called with the lock held."""
    if not state["ended"] and now_ms() >= state["deadline"]:
        state["ended"] = True
        state["ended_at"] = state["deadline"]
        save_state(state)


def empty_zone_counts():
    return {z["name"]: 0 for z in ZONES}


def player_achievements(pid):
    """Derive private achievements from recorded history, including legacy games."""
    quiet_target = max(1, state["duration_ms"] // 2)
    close_at = spectrum_at = quiet_at = None
    visited = set()
    longest_quiet = 0
    previous_at = state["started_at"]
    zone_names = {zone["name"] for zone in ZONES}
    extra_rules = [
        ("red_alert", "Red alert", "Press in the red zone.", "red", None),
        ("into_the_dark", "Into the dark", "Press in the black zone.", "black", None),
        ("white_knuckle", "White knuckle", "Press in the white zone.", "white", None),
        ("photo_finish", "Photo finish", "Press with less than ten seconds remaining.", None, 10000),
        ("last_second", "Last second", "Press with less than one second remaining.", None, 1000),
    ]
    extra_unlocks = {rule[0]: None for rule in extra_rules}
    for press in state["presses"]:
        quiet_ms = max(0, press["at"] - previous_at)
        previous_at = press["at"]  # measure the global gap, not this player's gap
        if press["player_id"] != pid:
            continue
        for achievement_id, _, _, zone, threshold in extra_rules:
            qualifies = press["zone"] == zone if zone is not None else 0 < press["remaining_ms"] < threshold
            if extra_unlocks[achievement_id] is None and qualifies:
                extra_unlocks[achievement_id] = press["at"]
        if close_at is None and 0 < press["remaining_ms"] < 60000:
            close_at = press["at"]
        if press["zone"] in zone_names:
            visited.add(press["zone"])
        if spectrum_at is None and visited == zone_names:
            spectrum_at = press["at"]
        longest_quiet = max(longest_quiet, quiet_ms)
        if quiet_at is None and quiet_ms >= quiet_target:
            quiet_at = press["at"]
    return [
        {"id": "close_call", "name": "Close call",
         "description": "Press with less than one minute remaining.",
         "unlocked": close_at is not None, "unlocked_at": close_at,
         "progress": int(close_at is not None), "target": 1},
        {"id": "full_spectrum", "name": "Full spectrum",
         "description": "Press in all six scoring zones.",
         "unlocked": spectrum_at is not None, "unlocked_at": spectrum_at,
         "progress": len(visited), "target": len(zone_names)},
        {"id": "keeping_us_alive", "name": "Keeping us alive",
         "description": "Press after nobody has pressed for at least half the game's countdown.",
         "unlocked": quiet_at is not None, "unlocked_at": quiet_at,
         "progress": min(longest_quiet, quiet_target), "target": quiet_target},
    ] + [
        {"id": achievement_id, "name": name, "description": description,
         "unlocked": extra_unlocks[achievement_id] is not None,
         "unlocked_at": extra_unlocks[achievement_id],
         "progress": int(extra_unlocks[achievement_id] is not None), "target": 1}
        for achievement_id, name, description, _, _ in extra_rules
    ]


def player_summary(pid):
    presses = [p for p in state["presses"] if p["player_id"] == pid]
    counts = empty_zone_counts()
    for p in presses:
        counts[p["zone"]] += 1
    standings = ranked_players()
    position = next(i for i, (player_id, _) in enumerate(standings) if player_id == pid)
    points = standings[position][1]["total_points"]
    rival = standings[position - 1][1] if position else None

    def record(press):
        return {key: press[key] for key in ("at", "remaining_ms", "zone", "points")} if press else None

    closest = min(presses, key=lambda p: (p["remaining_ms"], p["at"]), default=None)
    best = min(presses, key=lambda p: (-p["points"], p["remaining_ms"], p["at"]), default=None)
    return {
        "name": state["players"][pid]["name"],
        "subscribed": state["players"][pid].get("subscribed", False),
        "total_players": len(state["players"]),
        "total_presses": len(state["presses"]),
        "total_points": sum(p["points"] for p in presses),
        "press_count": len(presses),
        "zone_counts": counts,
        "best_remaining_ms": min((p["remaining_ms"] for p in presses), default=None),
        "personal_bests": {"closest_call": record(closest), "highest_scoring_press": record(best)},
        "achievements": player_achievements(pid),
        "ranking": {
            "position": position + 1,
            "total_players": len(standings),
            "tied_for_lead": len(standings) > 1 and points == standings[0][1]["total_points"]
                and sum(score["total_points"] == points for _, score in standings) > 1,
            "rival": {
                "name": rival["name"],
                "total_points": rival["total_points"],
                "points_to_tie": rival["total_points"] - points,
                "points_to_pass": rival["total_points"] - points + 1,
            } if rival else None,
        },
        "presses": [
            {k: p[k] for k in ("at", "remaining_ms", "zone", "points")} for p in presses
        ],
    }


def public_press(p):
    return {
        "name": state["players"][p["player_id"]]["name"],
        "at": p["at"],
        "remaining_ms": p["remaining_ms"],
        "zone": p["zone"],
        "points": p["points"],
    }


def ranked_players():
    """Internal standings retain IDs to identify players even in legacy data."""
    scores = {
        pid: {"name": player["name"], "total_points": 0, "press_count": 0}
        for pid, player in state["players"].items()
    }
    for press in state["presses"]:
        score = scores[press["player_id"]]
        score["total_points"] += press["points"]
        score["press_count"] += 1
    return sorted(scores.items(), key=lambda item: (-item[1]["total_points"], item[1]["name"].lower()))


def public_leaderboard():
    """Public score totals only; do not expose sessions or individual presses."""
    return [score for _, score in ranked_players()]


def global_stats():
    counts = empty_zone_counts()
    for p in state["presses"]:
        counts[p["zone"]] += 1
    leaderboard = []
    for pid in state["players"]:
        s = player_summary(pid)
        del s["presses"], s["subscribed"], s["personal_bests"], s["ranking"], s["achievements"]
        leaderboard.append(s)
    leaderboard.sort(key=lambda s: (-s["total_points"], s["name"].lower()))
    presses = state["presses"]
    lowest = min(presses, key=lambda p: p["remaining_ms"], default=None)
    return {
        "started_at": state["started_at"],
        "ended_at": state["ended_at"],
        "total_players": len(state["players"]),
        "total_presses": len(presses),
        "zone_counts": counts,
        "leaderboard": leaderboard,
        "last_press": public_press(presses[-1]) if presses else None,
        "lowest_press": public_press(lowest) if lowest else None,
        "presses": [public_press(p) for p in presses],
    }


class Handler(BaseHTTPRequestHandler):
    server_version = "TheButton/1.0"

    def log_message(self, fmt, *args):
        pass

    def send_json(self, status, payload):
        body = json.dumps(payload).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Cache-Control", "no-store")
        self.send_common_headers()
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def send_common_headers(self):
        self.send_header("X-Content-Type-Options", "nosniff")
        self.send_header("X-Frame-Options", "DENY")
        self.send_header("Referrer-Policy", "no-referrer")
        self.send_header(
            "Content-Security-Policy",
            "default-src 'self'; script-src 'self' https://js.stripe.com; "
            "frame-src 'self' https://js.stripe.com; frame-ancestors 'none'; "
            "base-uri 'none'; form-action 'self'",
        )

    def read_json(self):
        length = int(self.headers.get("Content-Length") or 0)
        if length > MAX_BODY_BYTES:
            raise ValueError("Request body too large")
        raw = self.rfile.read(length) if length else b"{}"
        data = json.loads(raw.decode("utf-8"))
        if not isinstance(data, dict):
            raise ValueError("Expected a JSON object")
        return data

    def bearer_hash(self):
        auth = self.headers.get("Authorization", "")
        if not auth.startswith("Bearer "):
            return None
        return hash_token(auth[7:].strip())

    def authed_player(self):
        th = self.bearer_hash()
        return token_index.get(th) if th else None

    def do_GET(self):
        path = self.path.split("?", 1)[0]
        if path == "/api/state":
            with lock:
                check_ended()
                return self.send_json(HTTPStatus.OK, {
                    "now": now_ms(),
                    "started_at": state["started_at"],
                    "deadline": state["deadline"],
                    "duration_ms": DURATION_MS,
                    "ended": state["ended"],
                    "total_players": len(state["players"]),
                    "total_presses": len(state["presses"]),
                    "zones": ZONES,
                })
        if path == "/api/leaderboard":
            with lock:
                return self.send_json(HTTPStatus.OK, {"leaderboard": public_leaderboard()})
        if path == "/api/chat":
            return self.send_json(HTTPStatus.OK, {"messages": chat_messages()})
        if path == "/api/presence":
            with lock:
                return self.send_json(HTTPStatus.OK, public_presence())
        if path == "/api/me":
            with lock:
                pid = self.authed_player()
                if pid is None:
                    return self.send_json(HTTPStatus.UNAUTHORIZED, {"error": "Not registered"})
                return self.send_json(HTTPStatus.OK, player_summary(pid))
        if path == "/api/stats":
            with lock:
                check_ended()
                if not state["ended"]:
                    return self.send_json(HTTPStatus.FORBIDDEN, {"error": "Stats are published when the game ends"})
                return self.send_json(HTTPStatus.OK, global_stats())
        if path in STATIC_FILES:
            filename, ctype = STATIC_FILES[path]
            body = (STATIC_DIR / filename).read_bytes()
            self.send_response(HTTPStatus.OK)
            self.send_header("Content-Type", ctype)
            self.send_common_headers()
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)
            return
        self.send_json(HTTPStatus.NOT_FOUND, {"error": "Not found"})

    def do_POST(self):
        path = self.path.split("?", 1)[0]
        try:
            data = self.read_json()
        except (ValueError, UnicodeDecodeError):
            return self.send_json(HTTPStatus.BAD_REQUEST, {"error": "Invalid JSON body"})

        if path == "/api/presence":
            with lock:
                th = self.bearer_hash()
                if th not in token_index:
                    return self.send_json(HTTPStatus.UNAUTHORIZED, {"error": "Log in to share your presence"})
                tab = data.get("tab_id")
                active, in_chat = data.get("active"), data.get("in_chat", False)
                if (not isinstance(tab, str) or re.fullmatch(r"[a-f0-9]{32}", tab) is None
                        or not isinstance(active, bool) or not isinstance(in_chat, bool)):
                    return self.send_json(HTTPStatus.BAD_REQUEST, {"error": "Invalid presence update"})
                prune_presence()
                key = (th, tab)
                if not active:
                    presence.pop(key, None)
                else:
                    if key not in presence and sum(k[0] == th for k in presence) >= MAX_PRESENCE_TABS_PER_SESSION:
                        return self.send_json(HTTPStatus.TOO_MANY_REQUESTS, {"error": "Too many active tabs"})
                    presence[key] = (time.monotonic(), in_chat)
                return self.send_json(HTTPStatus.OK, public_presence())

        if path == "/api/chat":
            with lock:
                pid = self.authed_player()
                if pid is None:
                    return self.send_json(HTTPStatus.UNAUTHORIZED, {"error": "Log in to send messages"})
                message = data.get("message")
                if not isinstance(message, str) or not (1 <= len(message.strip()) <= MAX_CHAT_LENGTH):
                    return self.send_json(HTTPStatus.BAD_REQUEST, {"error": f"Message must be 1-{MAX_CHAT_LENGTH} characters"})
                message = message.strip()
                if not message.isprintable():
                    return self.send_json(HTTPStatus.BAD_REQUEST, {"error": "Message contains invalid characters"})
                with database() as connection:
                    connection.execute("BEGIN IMMEDIATE")
                    t = now_ms()
                    last = connection.execute(
                        "SELECT at FROM chat_messages WHERE player_id = ? AND kind = 'message' ORDER BY id DESC LIMIT 1", (pid,),
                    ).fetchone()
                    if last is not None and t - last["at"] < CHAT_COOLDOWN_MS:
                        return self.send_json(HTTPStatus.TOO_MANY_REQUESTS, {"error": "Please wait two seconds between messages"})
                    name = state["players"][pid]["name"]
                    cursor = connection.execute(
                        "INSERT INTO chat_messages (player_id, name, at, message) VALUES (?, ?, ?, ?)",
                        (pid, name, t, message),
                    )
                    posted = {"id": cursor.lastrowid, "name": name, "at": t, "message": message, "kind": "message"}
            return self.send_json(HTTPStatus.CREATED, posted)

        if path == "/api/register":
            name, password, error = validate_credentials(data)
            if error:
                return self.send_json(HTTPStatus.BAD_REQUEST, {"error": error})
            password_hash = hash_password(password)  # slow; done outside the lock
            with lock:
                check_ended()
                if state["ended"]:
                    return self.send_json(HTTPStatus.CONFLICT, {"error": "The game is over"})
                if find_player_by_name(name) is not None:
                    return self.send_json(HTTPStatus.CONFLICT, {"error": "Name already taken"})
                pid = secrets.token_hex(8)
                state["players"][pid] = {
                    "name": name,
                    "password_hash": password_hash,
                    "token_hashes": [],
                    "created_at": now_ms(),
                }
                token = issue_token(pid)
            return self.send_json(HTTPStatus.CREATED, {"name": name, "token": token})

        if path == "/api/login":
            name, password = data.get("name"), data.get("password")
            if not isinstance(name, str) or not isinstance(password, str) or len(password) > MAX_PASSWORD_LEN:
                return self.send_json(HTTPStatus.BAD_REQUEST, {"error": "Name and password required"})
            name = " ".join(name.split())
            key = name.lower()
            with lock:
                count, first = failed_logins.get(key, (0, 0))
                if count >= LOGIN_MAX_FAILURES and now_ms() - first < LOGIN_LOCKOUT_MS:
                    return self.send_json(HTTPStatus.TOO_MANY_REQUESTS, {"error": "Too many failed attempts; try again later"})
                pid = find_player_by_name(name)
                stored = state["players"][pid].get("password_hash") if pid else None
            ok = verify_password(password, stored or DUMMY_PASSWORD_HASH) and stored is not None
            with lock:
                if not ok:
                    count, first = failed_logins.get(key, (0, 0))
                    if now_ms() - first >= LOGIN_LOCKOUT_MS:
                        count, first = 0, now_ms()
                    failed_logins[key] = (count + 1, first)
                    return self.send_json(HTTPStatus.UNAUTHORIZED, {"error": "Invalid name or password"})
                failed_logins.pop(key, None)
                token = issue_token(pid)
                return self.send_json(HTTPStatus.OK, {"name": state["players"][pid]["name"], "token": token})

        if path == "/api/logout":
            with lock:
                th = self.bearer_hash()
                pid = token_index.pop(th, None) if th else None
                prune_presence()
                if pid is not None:
                    state["players"][pid]["token_hashes"].remove(th)
                    save_state(state)
            return self.send_json(HTTPStatus.OK, {"ok": True})

        if path == "/api/subscribe":
            with lock:
                pid = self.authed_player()
                player_name = state["players"][pid]["name"] if pid else None
                already_subscribed = bool(pid and state["players"][pid].get("subscribed"))
            if pid is None:
                return self.send_json(HTTPStatus.UNAUTHORIZED, {"error": "Log in to join the mailing list"})
            if already_subscribed:
                return self.send_json(HTTPStatus.OK, {"ok": True})
            entry, error = validate_subscription(data)
            if error:
                return self.send_json(HTTPStatus.BAD_REQUEST, {"error": error})
            entry["player"] = player_name
            with mailing_lock:
                # Same response for repeat emails so the endpoint doesn't reveal who is subscribed.
                if entry["email"].lower() not in subscribed_emails:
                    add_subscriber(entry)
            with lock:
                state["players"][pid]["subscribed"] = True
                save_state(state)
            return self.send_json(HTTPStatus.OK, {"ok": True})

        if path == "/api/press":
            with lock:
                pid = self.authed_player()
                if pid is None:
                    return self.send_json(HTTPStatus.UNAUTHORIZED, {"error": "Not registered"})
                check_ended()
                if state["ended"]:
                    return self.send_json(HTTPStatus.CONFLICT, {"error": "The game is over"})
                t = now_ms()
                remaining = state["deadline"] - t
                zone = zone_for_remaining(remaining)
                press = {
                    "player_id": pid,
                    "at": t,
                    "remaining_ms": remaining,
                    "zone": zone["name"],
                    "points": zone["points"],
                }
                state["presses"].append(press)
                state["deadline"] = t + DURATION_MS
                hours, remainder = divmod(max(0, remaining), 3600000)
                minutes, remainder = divmod(remainder, 60000)
                seconds, milliseconds = divmod(remainder, 1000)
                time_left = f"{hours:02d}:{minutes:02d}:{seconds:02d}.{milliseconds:03d}"
                name = state["players"][pid]["name"]
                points = press["points"]
                announcement = f"{name} pushed the button with {time_left} remaining and received {points} {'point' if points == 1 else 'points'}."
                try:
                    with database() as connection:
                        write_state(connection, state)
                        connection.execute(
                            "INSERT INTO chat_messages (player_id, name, at, message, kind) VALUES (?, ?, ?, ?, 'press')",
                            (pid, "The Button", t, announcement),
                        )
                except Exception:
                    # Keep memory consistent with the rolled-back database snapshot.
                    state["presses"].pop()
                    state["deadline"] = t + remaining
                    raise
                return self.send_json(HTTPStatus.OK, {
                    "press": {k: press[k] for k in ("at", "remaining_ms", "zone", "points")},
                    "deadline": state["deadline"],
                    "now": t,
                })

        self.send_json(HTTPStatus.NOT_FOUND, {"error": "Not found"})


mailing_lock = threading.Lock()


def load_subscribed_emails():
    if not MAILING_LIST_FILE.exists():
        return set()
    with MAILING_LIST_FILE.open("r", newline="", encoding="utf-8") as f:
        return {row["email"].lower() for row in csv.DictReader(f)}


subscribed_emails = load_subscribed_emails()


def clean_text(value, max_len):
    """Returns a whitespace-normalized string, "" for missing values, or None if invalid."""
    if value is None:
        return ""
    if not isinstance(value, str):
        return None
    value = " ".join(value.split())
    if len(value) > max_len or not value.isprintable():
        return None
    return value


def validate_subscription(data):
    first = clean_text(data.get("first_name"), 50)
    last = clean_text(data.get("last_name"), 50)
    email = clean_text(data.get("email"), 254)
    phone = clean_text(data.get("phone"), 25)
    if not first:
        return None, "First name is required (max 50 characters)"
    if last is None:
        return None, "Last name must be at most 50 characters"
    if not email or not EMAIL_RE.match(email):
        return None, "A valid email address is required"
    if phone is None or (phone and not PHONE_RE.match(phone)):
        return None, "Phone number is invalid"
    return {"first_name": first, "last_name": last, "email": email, "phone": phone}, None


def csv_safe(value):
    # Stops spreadsheet apps from interpreting user input as a formula.
    return "'" + value if value[:1] in ("=", "+", "-", "@") else value


def add_subscriber(entry):
    """Must be called with mailing_lock held."""
    is_new_file = not MAILING_LIST_FILE.exists()
    row = {k: csv_safe(v) for k, v in entry.items()}
    row["subscribed_at"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
    with MAILING_LIST_FILE.open("a", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=MAILING_LIST_FIELDS)
        if is_new_file:
            writer.writeheader()
        writer.writerow(row)
    subscribed_emails.add(entry["email"].lower())


def end_watcher():
    while True:
        with lock:
            check_ended()
            if state["ended"]:
                return
            wait = (state["deadline"] - now_ms()) / 1000
        time.sleep(min(max(wait, 0.05), 60))


if __name__ == "__main__":
    threading.Thread(target=end_watcher, daemon=True).start()
    print(f"The Button running on http://{HOST}:{PORT}  (timer: {DURATION_MS / 1000:.0f}s)")
    ThreadingHTTPServer((HOST, PORT), Handler).serve_forever()