"""Local proof, not a production server or a test of a named hosting product.
Requires macOS, Python 3 and Segno 1.6.1. Keep decode-qr.swift beside this file.
All HTTP traffic is loopback. .test hostnames are injected in Host; no DNS/TLS.
"""
from pathlib import Path
import datetime, hashlib, http.client, http.server, json, platform, socket
import sqlite3, subprocess, threading
from urllib.parse import urlsplit
import segno

ROOT = Path(__file__).resolve().parent
OUT = ROOT / "experiment"
OUT.mkdir(exist_ok=True)
RUN = OUT / datetime.datetime.now(datetime.timezone.utc).strftime("%Y%m%dT%H%M%SZ")
RUN.mkdir()
HOST = "go.qrbench.test"
ITEMS = ["menu", "guide", "contact"]
urls = {s: f"http://{HOST}/{s}" for s in ITEMS}

def sha(path):
    return hashlib.sha256(path.read_bytes()).hexdigest()

# Fail closed for Python-level network connects while producing the files.
original_connect = socket.socket.connect
def blocked(*args, **kwargs):
    raise AssertionError("Network connection attempted during QR generation")
socket.socket.connect = blocked
try:
    for slug, url in urls.items():
        qr = segno.make(url, error="m", micro=False, boost_error=False)
        qr.save(RUN / f"{slug}.png", scale=12, border=4)
        qr.save(RUN / f"{slug}.svg", scale=8, border=4)
finally:
    socket.socket.connect = original_connect
hashes_before = {s: sha(RUN / f"{s}.png") for s in ITEMS}

def decode():
    decoder = ROOT / "decode-qr.swift"
    result = subprocess.run(["swift", "-module-cache-path", "/tmp/qrbench-selfhosting-swift-cache", str(decoder), *[str(RUN/f"{s}.png") for s in ITEMS]],
                            text=True, capture_output=True, check=True)
    for s, url in urls.items():
        assert f"{s}.png → ENCODE: {url}" in result.stdout, result.stdout
    return result.stdout

decoded_before = decode()

def database(path, records):
    with sqlite3.connect(path) as db:
        db.execute("CREATE TABLE IF NOT EXISTS routes(host TEXT, path TEXT, target TEXT, PRIMARY KEY(host,path))")
        db.executemany("INSERT OR REPLACE INTO routes VALUES(?,?,?)", records)

db_a = RUN / "source.sqlite"
db_b = RUN / "replacement.sqlite"
records = [(HOST, f"/{s}", f"http://destination.test/{s}-v1") for s in ITEMS]
database(db_a, records)
database(db_b, [])

def server(path):
    class Handler(http.server.BaseHTTPRequestHandler):
        def do_GET(self):
            host = self.headers.get("Host", "")
            if host == "destination.test" and self.path in [f"/{s}-v{v}" for s in ITEMS for v in (1,2)]:
                self.send_response(200)
                self.end_headers()
                self.wfile.write(self.path.encode())
                return
            with sqlite3.connect(path) as db:
                row = db.execute("SELECT target FROM routes WHERE host=? AND path=?", (host,self.path)).fetchone()
            self.send_response(302 if row else 404)
            if row:
                self.send_header("Location", row[0])
            self.send_header("Cache-Control", "no-store")
            self.end_headers()
        def log_message(self, *args):
            pass
    srv = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
    thread = threading.Thread(target=srv.serve_forever, daemon=True)
    thread.start()
    return srv, thread

def stop(srv, thread):
    srv.shutdown()
    srv.server_close()
    thread.join()

def request(port, url):
    parsed = urlsplit(url)
    assert parsed.scheme == "http" and parsed.hostname.endswith(".test")
    c = http.client.HTTPConnection("127.0.0.1", port, timeout=2)
    c.request("GET", parsed.path, headers={"Host":parsed.netloc})
    res = c.getresponse()
    item = {"url":url,"status":res.status,"location":res.getheader("Location"),"body":res.read().decode()}
    c.close()
    return item

def journey(port, url):
    first = request(port,url)
    if first["status"] == 302:
        first["destination"] = request(port, first["location"])
        assert first["destination"]["status"] == 200
    return first

a, ta = server(db_a)
port_a = a.server_port
steps = []
try:
    before = [journey(port_a,u) for u in urls.values()]
    assert all(x["status"] == 302 for x in before)
    steps.append({"stage":"source_running","results":before})
    with sqlite3.connect(db_a) as src, sqlite3.connect(RUN/"backup.sqlite") as dst:
        src.backup(dst)
finally:
    stop(a, ta)
try:
    request(port_a, urls["menu"])
    raise AssertionError("Stopped source still answers")
except (ConnectionRefusedError, ConnectionResetError):
    steps.append({"stage":"source_stopped","result":"connection_refused"})

b, tb = server(db_b)
try:
    empty = [journey(b.server_port,u) for u in urls.values()]
    assert all(x["status"] == 404 for x in empty)
    steps.append({"stage":"replacement_empty","results":empty})
    with sqlite3.connect(RUN/"backup.sqlite") as src, sqlite3.connect(db_b) as dst:
        src.backup(dst)
    restored = [journey(b.server_port,u) for u in urls.values()]
    assert [x["location"] for x in restored] == [x["location"] for x in before]
    steps.append({"stage":"replacement_restored","results":restored})
    with sqlite3.connect(db_b) as db:
        db.execute("UPDATE routes SET target=? WHERE path='/menu'", ("http://destination.test/menu-v2",))
    changed = journey(b.server_port,urls["menu"])
    assert changed["destination"]["body"] == "/menu-v2"
    steps.append({"stage":"destination_changed","result":changed})
    # Preserve the source route but recreate it under a different hostname.
    with sqlite3.connect(db_b) as db:
        db.execute("UPDATE routes SET host='new.qrbench.test'")
    old_host = [journey(b.server_port,u) for u in urls.values()]
    assert all(x["status"] == 404 for x in old_host)
    new_host = journey(b.server_port,urls["menu"].replace(HOST,"new.qrbench.test"))
    assert new_host["status"] == 302
    steps.append({"stage":"hostname_changed_original_qrs","results":old_host,"new_url_control":new_host})
finally:
    stop(b,tb)

decoded_after = decode()
hashes_after = {s: sha(RUN / f"{s}.png") for s in ITEMS}
assert hashes_before == hashes_after
result = {"performed_at":datetime.datetime.now(datetime.timezone.utc).isoformat(),
          "python":platform.python_version(),"segno":segno.__version__,
          "qr_count":3,"generation_socket_connect_blocked":True,
          "qr_hashes_before":hashes_before,"qr_hashes_after":hashes_after,
          "vision_before":decoded_before,"vision_after":decoded_after,"steps":steps,
          "limits":["Loopback HTTP only; Host headers explicitly routed by client, no public DNS or TLS tested.",
                    "Same laptop, two server instances and separate SQLite files, not two physical hosts.",
                    "No Shlink installation, no vendor account, no phone/camera or printed scan test.",
                    "Three controlled URLs; no speed, uptime, scalability or market-wide claim."]}
(RUN/"results.json").write_text(json.dumps(result,ensure_ascii=False,indent=2)+"\n")
(OUT/"latest.txt").write_text(RUN.name+"\n")
print(json.dumps({"run":str(RUN),"qr_count":3,"checks":"PASS","stages":[s["stage"] for s in steps]},ensure_ascii=False))