"""Local stub of an ODIAN site for the odian-packs Rust end-to-end tests.

    PYTHONPATH=<ODIAN_MCP_Server>/src python stub_site.py --root DIR [--no-empty-src]

Builds a small signed pack store under DIR with odian_packs itself. The layout is the
site store layout, which is also the ODIAN drive layout:

    DIR/packs/versions/CURRENT
    DIR/packs/versions/0.51.0/{manifest.json, notes.md, tiles/dark.test_ds.index.json}
    DIR/packs/app/<sha256>            DIR/packs/db/<snapshot>/...
    DIR/tiles/dark/test_ds/<z>/<x>/<y><ext>, DIR/tiles/tiles.db
    DIR/src/...

It then serves DIR on 127.0.0.1 (a free port) with ``odian_update.api.create_app`` and a
stub token validator (token ``good``), plus nginx-like Range routes for /packs/app,
/packs/db and /src that check the same Bearer token. It prints one JSON line (port, test
public key, token, layout) and serves until its stdin closes.

Test-only control routes (no token):
    GET  /_ctl/log                                   every request seen: path, range
    POST /_ctl/reset                                 clear the log and all rules
    POST /_ctl/fail?prefix=P&status=503&count=N&retry_after=0
    POST /_ctl/cut?prefix=P&after=BYTES&count=N      send BYTES of the body, then drop the connection
"""

import argparse
import hashlib
import json
import os
import re
import shutil
import socket
import sys
import threading
import urllib.parse

sys.dont_write_bytecode = True

VERSION = "0.51.0"
STYLE = "dark"
DATASET = "test_ds"
SNAPSHOT = "20260915T101500Z"
TOKEN = "good"
CHUNK = 1024 * 1024
TEST_KEY_RAW = bytes(range(32))

APP_FILES = {
    "Odian/Odian.exe": 200 * 1024,
    "Odian/assets/empty.conf": 0,
    "Odian/data/big.bin": 3 * CHUNK + 17,
    "Odian/données/carte été.png": 1000,
    "api/odian_api.exe": 50 * 1024,
}
DB_FILES = [
    ("pre-data", None, "pre-data.dump", 5000),
    ("data", "public.t1", "data/public.t1.dump", 300 * 1024),
    ("post-data", None, "post-data.dump", 2048),
]
# (z, x, y, ext, size). Blocks (0,0,0), (3,0,0), (7,0,0), (7,1,0). The 8-byte tile is
# below MIN_TILE_BYTES and never journaled.
TILES = [
    (0, 0, 0, ".png", 5000),
    (3, 1, 2, ".png", 40),
    (7, 1, 2, ".png", 700),
    (7, 3, 2, ".jpg", 512),
    (7, 3, 2, ".png", 9),
    (7, 63, 63, ".pbf", 300 * 1024),
    (7, 2, 2, ".png", 8),
    (7, 64, 5, ".png", 20),
    (7, 127, 63, ".png", 1000),
]
# Journaled AFTER the index is frozen, inside block (7,0,0): no install may contain it.
LATE_TILE = (7, 5, 5, ".png", 64)
SRC_FILES = {"osm_source/world_aeroway_v1.sqlite": 2 * CHUNK + 123, "osm_source/small.bin": 100}
EMPTY_SRC = "osm_source/empty.bin"


def blob(label, size):
    return hashlib.shake_256(label.encode("utf-8")).digest(size) if size else b""


def write(path, data):
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(path, "wb") as f:
        f.write(data)


def write_json(path, obj):
    """Like odian_packs.cli._write_json: pretty, sorted, ASCII, trailing newline."""
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(path, "w", encoding="utf-8", newline="\n") as f:
        json.dump(obj, f, indent=2, sort_keys=True)
        f.write("\n")


def join(root, rel):
    return os.path.join(root, *rel.split("/"))


def build(root, empty_src):
    from odian_packs import appstore, hashing, signing, tiles
    from odian_packs import manifest as mf

    packs = os.path.join(root, "packs")
    tiles_root = os.path.join(root, "tiles")
    src_root = os.path.join(root, "src")
    stage = os.path.join(root, "stage")

    for rel, size in APP_FILES.items():
        write(join(os.path.join(stage, "app"), rel), blob("app:" + rel, size))
    app = appstore.store_tree(os.path.join(stage, "app"), packs, workers=2)
    shutil.rmtree(stage)

    db_files = []
    for section, table, rel, size in DB_FILES:
        path_rel = f"db/{SNAPSHOT}/{rel}"
        full = join(packs, path_rel)
        write(full, blob("db:" + rel, size))
        got_size, got_sha = hashing.sha256_file(full)
        entry = {"section": section, "path": path_rel, "size": got_size, "sha256": got_sha}
        if table:
            entry["table"] = table
        db_files.append(entry)

    set_root = os.path.join(tiles_root, STYLE, DATASET)
    journal = os.path.join(tiles_root, "tiles.db")
    for z, x, y, ext, size in TILES:
        write(os.path.join(set_root, str(z), str(x), f"{y}{ext}"), blob(f"tile:{z}/{x}/{y}{ext}", size))
    tiles.import_tree(set_root, STYLE, DATASET, journal, workers=2)
    index = tiles.build_index(journal, STYLE, DATASET)
    version_dir = os.path.join(packs, "versions", VERSION)
    index_name = mf.tile_index_name(STYLE, DATASET)
    write_json(join(version_dir, index_name), index)
    z, x, y, ext, size = LATE_TILE
    write(os.path.join(set_root, str(z), str(x), f"{y}{ext}"), blob(f"tile:{z}/{x}/{y}{ext}", size))
    tiles.import_tree(set_root, STYLE, DATASET, journal, workers=2)

    src_items = dict(SRC_FILES)
    if empty_src:
        src_items[EMPTY_SRC] = 0
    src_files = []
    for rel, size in src_items.items():
        full = join(src_root, rel)
        write(full, blob("src:" + rel, size))
        hashed = hashing.hash_chunks(full, CHUNK)
        src_files.append({"path": rel, "size": hashed["size"], "chunks": hashed["chunks"],
                          "chunks_sha256": hashed["chunks_sha256"]})

    manifest = {
        "schema": 1, "product": "odian", "version": VERSION, "released_at": "2026-09-15T12:00:00Z",
        "min_launcher": "0.1.0", "notes": "notes.md",
        "app": app,
        "db": {"schema_version": 0, "snapshot": SNAPSHOT, "files": db_files},
        "tiles": {"block_shift": 6, "sets": [{
            "style": STYLE, "dataset": DATASET, "journal_seq": index["journal_seq"],
            "blocks": len(index["blocks"]), "files": index["files"], "bytes": index["bytes"],
            "index": index_name, "index_sha256": tiles.index_sha256(index)}]},
        "src": {"enabled": True, "chunk_bytes": CHUNK, "files": src_files},
    }
    signed = signing.sign(mf.validate(manifest), signing.private_key_from_raw(TEST_KEY_RAW))
    write_json(os.path.join(version_dir, "manifest.json"), signed)
    write(os.path.join(version_dir, "notes.md"), "# ODIAN 0.51.0\n\nPremier pack signé.\n".encode("utf-8"))
    write(os.path.join(packs, "versions", "CURRENT"), f"{VERSION}\n".encode("ascii"))


def serve(root, info):
    import uvicorn
    from fastapi import Request
    from fastapi.responses import Response

    from odian_packs.manifest import ManifestError, safe_relpath
    from odian_update.api import create_app
    from odian_update.auth import AuthInvalid, Caller, bearer_token
    from odian_update.config import Settings

    class StubValidator:
        async def validate(self, token):
            if token != TOKEN:
                raise AuthInvalid("invalid token")
            return Caller(username="stub-operator", expires_at=0)

    packs = os.path.join(root, "packs")
    settings = Settings(packs_dir=packs, tiles_dir=os.path.join(root, "tiles"), max_streams=8)
    app = create_app(settings, validator=StubValidator())

    def serve_file(request: Request, base: str, rel: str) -> Response:
        """nginx `alias` + `auth_request`: token check, then the file, honouring one Range."""
        if bearer_token(request.headers.get("authorization")) != TOKEN:
            return Response(status_code=401, headers={"WWW-Authenticate": "Bearer"})
        try:
            safe_relpath(rel)
        except ManifestError:
            return Response(status_code=404)
        path = join(base, rel)
        if not os.path.isfile(path):
            return Response(status_code=404)
        with open(path, "rb") as f:
            data = f.read()
        size = len(data)
        header = request.headers.get("range")
        if not header:
            return Response(data, media_type="application/octet-stream", headers={"Accept-Ranges": "bytes"})
        match = re.fullmatch(r"bytes=(\d*)-(\d*)", header.strip())
        unsatisfiable = Response(status_code=416, headers={"Content-Range": f"bytes */{size}"})
        if not match or (not match.group(1) and not match.group(2)):
            return unsatisfiable
        if not match.group(1):
            start, end = max(0, size - int(match.group(2))), size - 1
        else:
            start = int(match.group(1))
            end = min(int(match.group(2)), size - 1) if match.group(2) else size - 1
        if start >= size or start > end:
            return unsatisfiable
        return Response(data[start:end + 1], status_code=206, media_type="application/octet-stream",
                        headers={"Content-Range": f"bytes {start}-{end}/{size}", "Accept-Ranges": "bytes"})

    @app.get("/packs/app/{sha}")
    def app_file(sha: str, request: Request):
        return serve_file(request, os.path.join(packs, "app"), sha)

    @app.get("/packs/db/{rel:path}")
    def db_file(rel: str, request: Request):
        return serve_file(request, os.path.join(packs, "db"), rel)

    @app.get("/src/{rel:path}")
    def src_file(rel: str, request: Request):
        return serve_file(request, os.path.join(root, "src"), rel)

    lock = threading.Lock()
    state = {"log": [], "fail": [], "cut": []}

    async def reply(send, status, obj, headers=()):
        body = json.dumps(obj).encode("utf-8")
        await send({"type": "http.response.start", "status": status,
                    "headers": [(b"content-type", b"application/json"),
                                (b"content-length", str(len(body)).encode("ascii")), *headers]})
        await send({"type": "http.response.body", "body": body})

    async def control(scope, receive, send):
        path = scope["path"]
        query = {k: v[0] for k, v in urllib.parse.parse_qs(scope["query_string"].decode("latin-1")).items()}
        with lock:
            if path == "/_ctl/log":
                return await reply(send, 200, {"requests": list(state["log"])})
            if path == "/_ctl/reset":
                state["log"].clear()
                state["fail"].clear()
                state["cut"].clear()
            elif path == "/_ctl/fail":
                state["fail"].append({"prefix": query["prefix"], "status": int(query.get("status", "503")),
                                      "count": int(query.get("count", "1")), "retry_after": query.get("retry_after")})
            elif path == "/_ctl/cut":
                state["cut"].append({"prefix": query["prefix"], "after": int(query["after"]),
                                     "count": int(query.get("count", "1"))})
            else:
                return await reply(send, 404, {"error": "unknown control"})
        await reply(send, 200, {"ok": True})

    def take_rule(kind, path):
        for rule in state[kind]:
            if rule["count"] > 0 and path.startswith(rule["prefix"]):
                rule["count"] -= 1
                return rule
        return None

    async def asgi(scope, receive, send):
        if scope["type"] != "http":
            return await app(scope, receive, send)
        path = scope["path"]
        if path.startswith("/_ctl/"):
            return await control(scope, receive, send)
        headers = {k.decode("latin-1").lower(): v.decode("latin-1") for k, v in scope["headers"]}
        with lock:
            state["log"].append({"path": path, "range": headers.get("range")})
            fail = take_rule("fail", path)
            cut = None if fail else take_rule("cut", path)
        if fail:
            extra = [(b"retry-after", fail["retry_after"].encode("ascii"))] if fail["retry_after"] is not None else []
            return await reply(send, fail["status"], {"error": "stub failure"}, extra)
        if not cut:
            return await app(scope, receive, send)

        sent = 0

        async def cutting_send(message):
            nonlocal sent
            if message["type"] != "http.response.body":
                return await send(message)
            body = message.get("body", b"")
            room = cut["after"] - sent
            if len(body) < room:
                sent += len(body)
                return await send(message)
            await send({"type": "http.response.body", "body": body[:room], "more_body": True})
            raise ConnectionAbortedError("stub cut the connection")

        return await app(scope, receive, cutting_send)

    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.bind(("127.0.0.1", 0))
    port = sock.getsockname()[1]
    info.update(port=port, base_url=f"http://127.0.0.1:{port}")
    print(json.dumps(info), flush=True)

    def exit_when_stdin_closes():
        try:
            sys.stdin.buffer.read()
        finally:
            os._exit(0)

    threading.Thread(target=exit_when_stdin_closes, daemon=True).start()
    server = uvicorn.Server(uvicorn.Config(asgi, log_level="warning", lifespan="off", access_log=False))
    server.run(sockets=[sock])


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--root", required=True)
    parser.add_argument("--no-empty-src", action="store_true", help="leave out the zero-byte source file")
    parser.add_argument("--build-only", action="store_true")
    args = parser.parse_args()
    root = os.path.abspath(args.root)
    if not os.path.isfile(os.path.join(root, "packs", "versions", "CURRENT")):
        build(root, empty_src=not args.no_empty_src)

    from odian_packs import signing

    info = {
        "root": root, "token": TOKEN, "version": VERSION, "style": STYLE, "dataset": DATASET, "snapshot": SNAPSHOT,
        "public_key": signing.public_key_text(signing.private_key_from_raw(TEST_KEY_RAW).public_key()),
    }
    if args.build_only:
        print(json.dumps(info), flush=True)
        return
    serve(root, info)


if __name__ == "__main__":
    main()