#!/usr/bin/env python3
from __future__ import annotations

import base64
import concurrent.futures
import json
import os
from pathlib import Path
import random
import re
import shutil
import signal
import socket
import subprocess
import tempfile
import time
import urllib.error
import urllib.request
import uuid

ROOT = Path(__file__).resolve().parents[1]
API_SOURCE = ROOT / "achievement_api.php"


def achievement_ids() -> list[str]:
    source = API_SOURCE.read_text(encoding="utf-8")
    match = re.search(r"const ACHIEVEMENT_IDS = \[(.*?)\n\];", source, re.S)
    if not match:
        raise AssertionError("ACHIEVEMENT_IDS was not found")
    values = re.findall(r"'([^']+)'", match.group(1))
    if len(values) != 64 or len(values) != len(set(values)):
        raise AssertionError(f"expected 64 unique achievement IDs, got {len(values)}")
    return values


ACHIEVEMENTS = achievement_ids()


def compact_uuid(value: str) -> str:
    return base64.urlsafe_b64encode(uuid.UUID(value).bytes).decode("ascii").rstrip("=")


def expand_uuid(value: str) -> str:
    return str(uuid.UUID(bytes=base64.urlsafe_b64decode(value + "==")))


def decode_v3(value: object) -> dict:
    if not isinstance(value, list) or len(value) < 5 or value[0] != 3:
        raise AssertionError("state is not compact v3")
    base_time = int(value[1])
    versions = value[2]
    rows = value[3]
    unlock_rows = value[4]
    if not isinstance(versions, list) or not isinstance(rows, list) or not isinstance(unlock_rows, list):
        raise AssertionError("invalid v3 sections")
    players: dict[str, dict] = {}
    player_ids: list[str] = []
    for row in rows:
        player_id = expand_uuid(str(row[0]))
        first_seen = base_time + max(0, int(row[1] if len(row) > 1 else 0))
        last_seen = first_seen + max(0, int(row[2] if len(row) > 2 else 0))
        version_index = max(0, int(row[3] if len(row) > 3 else 0))
        player_ids.append(player_id)
        players[player_id] = {
            "firstSeen": first_seen,
            "lastSeen": last_seen,
            "gameVersion": str(versions[version_index] if version_index < len(versions) else ""),
        }
    unlocks: dict[str, dict[str, int]] = {}
    for row in unlock_rows:
        achievement_index = int(row[0])
        if achievement_index < 0 or achievement_index >= len(ACHIEVEMENTS):
            continue
        records: dict[str, int] = {}
        for offset in range(1, len(row) - 1, 2):
            player_index = int(row[offset])
            if 0 <= player_index < len(player_ids):
                records[player_ids[player_index]] = base_time + max(0, int(row[offset + 1]))
        if records:
            unlocks[ACHIEVEMENTS[achievement_index]] = records
    return {"players": players, "unlocks": unlocks}


def make_random_v2(seed: int, player_count: int, unlock_probability: float) -> tuple[dict, dict]:
    rng = random.Random(seed)
    versions = ["39.16.60", "39.16.63", "39.16.64", "39.16.65", "dev"]
    base_time = 1_780_000_000 + seed * 10_000
    rows: list[list] = []
    players: dict[str, dict] = {}
    player_ids: list[str] = []
    for index in range(player_count):
        player_id = str(uuid.UUID(int=rng.getrandbits(128), version=4))
        first_seen = base_time + rng.randint(0, 200_000)
        last_seen = first_seen + rng.randint(0, 80_000)
        version = rng.choices(versions, weights=[2, 4, 9, 12, 1], k=1)[0]
        player_ids.append(player_id)
        rows.append([player_id, first_seen, last_seen, version])
        players[player_id] = {"firstSeen": first_seen, "lastSeen": last_seen, "gameVersion": version}
    unlocks_flat: dict[str, list[int]] = {}
    unlocks: dict[str, dict[str, int]] = {}
    for achievement_id in ACHIEVEMENTS:
        flat: list[int] = []
        records: dict[str, int] = {}
        for player_index, player_id in enumerate(player_ids):
            if rng.random() >= unlock_probability:
                continue
            timestamp = players[player_id]["firstSeen"] + rng.randint(0, max(0, players[player_id]["lastSeen"] - players[player_id]["firstSeen"]))
            flat.extend([player_index, timestamp])
            records[player_id] = timestamp
        if flat:
            unlocks_flat[achievement_id] = flat
            unlocks[achievement_id] = records
    return {"v": 2, "p": rows, "u": unlocks_flat}, {"players": players, "unlocks": unlocks}


def make_random_verbose(seed: int, player_count: int, unlock_probability: float) -> tuple[dict, dict]:
    v2, expected = make_random_v2(seed, player_count, unlock_probability)
    players = {
        row[0]: {"firstSeen": row[1], "lastSeen": row[2], "gameVersion": row[3]}
        for row in v2["p"]
    }
    player_ids = [row[0] for row in v2["p"]]
    unlocks = {
        achievement_id: {player_ids[flat[i]]: flat[i + 1] for i in range(0, len(flat), 2)}
        for achievement_id, flat in v2["u"].items()
    }
    return {"players": players, "unlocks": unlocks}, expected


def normalize_expected(value: dict) -> dict:
    return {
        "players": {key: value["players"][key] for key in sorted(value["players"])},
        "unlocks": {
            achievement_id: {key: records[key] for key in sorted(records)}
            for achievement_id, records in value["unlocks"].items()
            if records
        },
    }


def free_port() -> int:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        sock.bind(("127.0.0.1", 0))
        return int(sock.getsockname()[1])


class PhpServer:
    def __init__(self, root: Path):
        self.root = root
        self.port = free_port()
        env = dict(os.environ)
        env["PHP_CLI_SERVER_WORKERS"] = "8"
        self.process = subprocess.Popen(
            ["php", "-S", f"127.0.0.1:{self.port}"],
            cwd=root,
            env=env,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            start_new_session=True,
        )
        deadline = time.time() + 8
        while time.time() < deadline:
            try:
                with socket.create_connection(("127.0.0.1", self.port), timeout=0.2):
                    break
            except OSError:
                if self.process.poll() is not None:
                    raise RuntimeError("PHP test server exited")
                time.sleep(0.05)
        else:
            raise RuntimeError("PHP test server did not start")

    def close(self) -> None:
        if self.process.poll() is None:
            os.killpg(self.process.pid, signal.SIGTERM)
            try:
                self.process.wait(timeout=4)
            except subprocess.TimeoutExpired:
                os.killpg(self.process.pid, signal.SIGKILL)
                self.process.wait(timeout=4)

    def request(self, *, action: str = "summary", player_id: str, payload: dict | None = None) -> tuple[int, dict]:
        if payload is None:
            url = f"http://127.0.0.1:{self.port}/achievement_api.php?action={action}&playerId={player_id}"
            request = urllib.request.Request(url, method="GET")
        else:
            url = f"http://127.0.0.1:{self.port}/achievement_api.php"
            data = json.dumps(payload, separators=(",", ":")).encode()
            request = urllib.request.Request(url, data=data, headers={"Content-Type": "application/json"}, method="POST")
        try:
            with urllib.request.urlopen(request, timeout=10) as response:
                return response.status, json.loads(response.read())
        except urllib.error.HTTPError as error:
            return error.code, json.loads(error.read())


def assert_equal_state(actual: dict, expected: dict, label: str) -> None:
    actual_normalized = normalize_expected(actual)
    expected_normalized = normalize_expected(expected)
    if actual_normalized != expected_normalized:
        raise AssertionError(f"{label}: semantic state changed")


def run_migration_cases(server: PhpServer, state_path: Path) -> tuple[int, int]:
    total_original = 0
    total_compact = 0
    probe_id = str(uuid.uuid4())
    for case_index in range(32):
        player_count = 1 + (case_index * 37) % 260
        probability = 0.02 + ((case_index * 11) % 55) / 100
        if case_index % 5 == 0:
            source, expected = make_random_verbose(10_000 + case_index, player_count, probability)
        else:
            source, expected = make_random_v2(10_000 + case_index, player_count, probability)
        original = json.dumps(source, separators=(",", ":")).encode()
        state_path.write_bytes(original)
        status, summary = server.request(player_id=probe_id)
        if status != 200 or not summary.get("ok"):
            raise AssertionError(f"migration case {case_index} failed: {status} {summary}")
        if summary.get("totalPlayers") != player_count:
            raise AssertionError("read-only summary changed or miscounted players")
        first_bytes = state_path.read_bytes()
        decoded = decode_v3(json.loads(first_bytes))
        assert_equal_state(decoded, expected, f"migration case {case_index}")
        status, second_summary = server.request(player_id=probe_id)
        if status != 200 or second_summary.get("totalPlayers") != player_count:
            raise AssertionError("v3 reread failed")
        if state_path.read_bytes() != first_bytes:
            raise AssertionError("canonical v3 changed on a second read")
        total_original += len(original)
        total_compact += len(first_bytes)
    if total_compact >= total_original * 0.72:
        raise AssertionError(f"aggregate compaction is insufficient: {total_compact}/{total_original}")
    return total_original, total_compact


def run_corruption_cases(server: PhpServer, state_path: Path) -> None:
    probe_id = str(uuid.uuid4())
    for payload in [b'{"v":2,', b'{"v":99}', b'[3,1]']:
        state_path.write_bytes(payload)
        status, body = server.request(player_id=probe_id)
        if status != 503 or body.get("error") != "storage_corrupt":
            raise AssertionError(f"corrupt state was not rejected: {status} {body}")
        if state_path.read_bytes() != payload:
            raise AssertionError("corrupt state was overwritten")


def run_completionist_case(server: PhpServer, state_path: Path) -> None:
    state_path.write_text("", encoding="utf-8")
    player_id = str(uuid.uuid4())
    now_ms = int(time.time() * 1000)
    status, body = server.request(player_id=player_id, payload={
        "action": "sync",
        "playerId": player_id,
        "gameVersion": "39.16.65",
        "completionistEligible": False,
        "unlocked": {"first_birth": now_ms, "true_tarinai_observer": now_ms},
    })
    if status != 200 or body["achievements"]["first_birth"]["unlockedPlayers"] != 1:
        raise AssertionError("ordinary sync unlock was lost")
    if body["achievements"]["true_tarinai_observer"]["unlockedPlayers"] != 0:
        raise AssertionError("ineligible completionist unlock was re-added")
    status, body = server.request(player_id=player_id, payload={
        "action": "sync",
        "playerId": player_id,
        "gameVersion": "39.16.65",
        "completionistEligible": True,
        "unlocked": {"true_tarinai_observer": now_ms},
    })
    if status != 200 or body["achievements"]["true_tarinai_observer"]["unlockedPlayers"] != 1:
        raise AssertionError("eligible completionist unlock was not accepted")
    before = state_path.read_bytes()
    status, body = server.request(action="reset", player_id=player_id)
    if status != 405 or body.get("error") != "method_not_allowed":
        raise AssertionError("GET reset was not rejected")
    if state_path.read_bytes() != before:
        raise AssertionError("rejected GET reset mutated state")


def run_concurrency_case(server: PhpServer, state_path: Path) -> None:
    state_path.write_text("", encoding="utf-8")
    rng = random.Random(52_001)
    requests: list[tuple[str, dict[str, int]]] = []
    now_ms = int(time.time() * 1000)
    for index in range(72):
        player_id = str(uuid.UUID(int=rng.getrandbits(128), version=4))
        selected = rng.sample(ACHIEVEMENTS, rng.randint(1, 18))
        unlocks = {achievement_id: now_ms - rng.randint(0, 1_000_000) for achievement_id in selected}
        requests.append((player_id, unlocks))

    def submit(entry: tuple[str, dict[str, int]]) -> None:
        player_id, unlocks = entry
        status, body = server.request(player_id=player_id, payload={
            "action": "sync",
            "playerId": player_id,
            "gameVersion": "39.16.65",
            "completionistEligible": "true_tarinai_observer" in unlocks,
            "unlocked": unlocks,
        })
        if status != 200 or not body.get("ok"):
            raise AssertionError(f"concurrent sync failed: {status} {body}")

    with concurrent.futures.ThreadPoolExecutor(max_workers=24) as executor:
        list(executor.map(submit, requests))

    status, summary = server.request(player_id=str(uuid.uuid4()))
    if status != 200 or summary.get("totalPlayers") != len(requests):
        raise AssertionError("concurrent updates lost players")
    decoded = decode_v3(json.loads(state_path.read_bytes()))
    if len(decoded["players"]) != len(requests):
        raise AssertionError("concurrent state player count mismatch")
    for player_id, unlocks in requests:
        for achievement_id in unlocks:
            if player_id not in decoded["unlocks"].get(achievement_id, {}):
                raise AssertionError(f"concurrent update lost {achievement_id} for {player_id}")


def main() -> None:
    with tempfile.TemporaryDirectory(prefix="tarinai-achievement-random-audit-") as temp_dir:
        root = Path(temp_dir)
        shutil.copy2(API_SOURCE, root / "achievement_api.php")
        data_dir = root / ".achievement_data"
        data_dir.mkdir()
        state_path = data_dir / "state.json"
        server = PhpServer(root)
        try:
            original, compact = run_migration_cases(server, state_path)
            run_corruption_cases(server, state_path)
            run_completionist_case(server, state_path)
            run_concurrency_case(server, state_path)
        finally:
            server.close()
    print(
        "[OK] 32 randomized v2/verbose migrations, exact semantic round-trips, corruption preservation, "
        f"completionist filtering, GET mutation rejection, and 72 concurrent syncs passed; "
        f"aggregate JSON shrank {original}→{compact} bytes ({compact / original:.1%})"
    )


if __name__ == "__main__":
    main()