"""End-to-end demo: the orders API and a partner's webhook receiver on real ports, driven over HTTP.

    python demo.py

Uses a throwaway SQLite file and random secrets; nothing to install beyond requirements.txt.
"""
import asyncio
import json
import os
import secrets
import shutil
import signal
import socket
import sqlite3
import subprocess
import sys
import tempfile
import time
from pathlib import Path

import httpx

HERE = Path(__file__).resolve().parent
PASSWORD = "brew-demo-pass"                        # the seeded demo accounts (dev only)


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


def start(target: str, port: int, env: dict, log_file: Path, *extra: str) -> subprocess.Popen:
    flags = subprocess.CREATE_NEW_PROCESS_GROUP if os.name == "nt" else 0
    log = log_file.open("w", encoding="utf-8")
    process = subprocess.Popen([sys.executable, "-m", "uvicorn", target, "--port", str(port), "--no-access-log",
                                "--timeout-graceful-shutdown", "10", *extra],
                               cwd=HERE, env=env, stdout=log, stderr=subprocess.STDOUT, creationflags=flags)
    process.log = log
    return process


def wait_until_up(url: str) -> None:
    for _ in range(150):
        try:
            if httpx.get(url, timeout=1).status_code < 500:
                return
        except httpx.TransportError:
            pass
        time.sleep(0.2)
    raise RuntimeError(f"{url} did not come up")


def stop(process: subprocess.Popen) -> None:
    process.send_signal(signal.CTRL_BREAK_EVENT if os.name == "nt" else signal.SIGTERM)
    process.wait(timeout=30)
    process.log.close()


def query(db_file: Path, sql: str) -> list[tuple]:
    connection = sqlite3.connect(db_file)             # read the API's database directly, then let it go
    try:
        return connection.execute(sql).fetchall()
    finally:
        connection.close()


def rupees(minor: int) -> str:
    return f"₹{minor / 100:,.2f}"


def heading(text: str) -> None:
    print(f"\n== {text}")


async def main() -> None:
    work = Path(tempfile.mkdtemp(prefix="orders-demo-"))
    db_file = work / "demo.db"
    receiver_port, api_port = free_port(), free_port()
    base_env = os.environ | {"ORDERS_WEBHOOK_SECRET": secrets.token_urlsafe(32), "PYTHONIOENCODING": "utf-8"}
    api_env = base_env | {
        "ORDERS_DATABASE_URL": f"sqlite+aiosqlite:///{db_file.as_posix()}",
        "ORDERS_JWT_SECRET": secrets.token_urlsafe(48),
        "ORDERS_WEBHOOK_URL": f"http://127.0.0.1:{receiver_port}/webhooks",
        "ORDERS_OUTBOX_POLL_SECONDS": "0.2",
    }
    receiver = start("demo_receiver:app", receiver_port, base_env, work / "receiver.log")
    api = start("orders_api.main:create_app", api_port, api_env, work / "api.log", "--factory")
    api_url = f"http://127.0.0.1:{api_port}"
    try:
        wait_until_up(f"http://127.0.0.1:{receiver_port}/received")
        wait_until_up(f"{api_url}/livez")
        async with httpx.AsyncClient(base_url=api_url, timeout=10) as http:
            await run(http, receiver_port, db_file)
    finally:
        stop(api)
        stop(receiver)
    entries = [json.loads(line) for line in (work / "api.log").read_text(encoding="utf-8").splitlines()
               if line.startswith("{")]
    heading("12. The API's JSON logs: one request by its id, then the graceful shutdown")
    for entry in entries:
        if entry["request_id"] == "demo-patch-0001":
            print("  " + json.dumps({k: v for k, v in entry.items() if k != "ts"}))
    for entry in entries[-4:]:
        print(f"  {entry['level']:<5} {entry['logger']:<14} {entry['msg']}")
    shutil.rmtree(work, ignore_errors=True)            # the throwaway database and logs


async def run(http: httpx.AsyncClient, receiver_port: int, db_file: Path) -> None:
    heading("1. Started: API and webhook receiver on real ports")
    print("  GET /readyz ->", (await http.get("/readyz")).json())

    heading("2. Log in (OAuth2 password flow, JWT access tokens)")
    tokens = {}
    for name in ("asha", "ravi", "admin"):
        t = time.perf_counter()
        r = await http.post("/v1/auth/token", data={"username": f"{name}@example.com", "password": PASSWORD})
        tokens[name] = {"Authorization": f"Bearer {r.json()['access_token']}"}
        me = (await http.get("/v1/me", headers=tokens[name])).json()
        print(f"  {me['email']:<18} role={me['role']:<8} token {r.json()['access_token'][:16]}…  "
              f"expires in {r.json()['expires_in']} s  (bcrypt check: {time.perf_counter() - t:.2f} s)")
    asha, ravi, admin = tokens["asha"], tokens["ravi"], tokens["admin"]

    heading("3. Place an order with an Idempotency-Key, then retry it")
    menu = await http.get("/v1/products")
    print(f"  menu: {len(menu.json())} products, Cache-Control: {menu.headers['cache-control']}")
    body = {"items": [{"sku": "latte", "quantity": 2}, {"sku": "croissant", "quantity": 1}], "note": "oat milk"}
    first = await http.post("/v1/orders", json=body, headers=asha | {"Idempotency-Key": "checkout-7f3a"})
    order = first.json()
    print(f"  POST /v1/orders       -> {first.status_code}  Location {first.headers['location']}  "
          f"ETag {first.headers['etag']}  total {rupees(order['totalMinor'])}")
    retry = await http.post("/v1/orders", json=body, headers=asha | {"Idempotency-Key": "checkout-7f3a"})
    print(f"  network retry, same key -> {retry.status_code}  Idempotent-Replayed: "
          f"{retry.headers.get('idempotent-replayed')}  same order: {retry.json()['id'] == order['id']}")
    clicks = await asyncio.gather(*(http.post("/v1/orders", json={"items": [{"sku": "mocha", "quantity": 1}]},
                                              headers=asha | {"Idempotency-Key": "double-click-19"})
                                    for _ in range(5)))
    print(f"  5 simultaneous clicks   -> statuses {sorted({c.status_code for c in clicks})}, "
          f"distinct orders created: {len({c.json()['id'] for c in clicks})}")
    reused = await http.post("/v1/orders", json={"items": [{"sku": "espresso", "quantity": 9}]},
                             headers=asha | {"Idempotency-Key": "checkout-7f3a"})
    print(f"  same key, other body    -> {reused.status_code}  {reused.json()['type'].rsplit('/', 1)[1]}")

    heading("4. Object-level authorization")
    url = f"/v1/orders/{order['id']}"
    print(f"  ravi GET asha's order          -> {(await http.get(url, headers=ravi)).status_code} (not 403: no hint it exists)")
    r = await http.post(f"{url}/status", json={"status": "completed"}, headers=ravi | {"If-Match": '"1.1"'})
    print(f"  ravi sets a status (admin only) -> {r.status_code}")
    r = await http.post("/v1/orders", json={"items": [{"sku": "latte", "quantity": 1}], "totalMinor": 1},
                        headers=asha | {"Idempotency-Key": "cheap-latte"})
    print(f"  asha sends her own price        -> {r.status_code} {r.json()['errors'][0]['msg']}: "
          f"{r.json()['errors'][0]['loc'][-1]}")

    heading("5. Conditional requests: ETag and If-Match")
    etag = (await http.get(url, headers=asha)).headers["etag"]
    print(f"  GET with If-None-Match {etag}  -> {(await http.get(url, headers=asha | {'If-None-Match': etag})).status_code}")
    print(f"  PATCH without If-Match      -> {(await http.patch(url, json={'note': 'soy'}, headers=asha)).status_code}")
    ok = await http.patch(url, json={"note": "oat milk, extra hot"},
                          headers=asha | {"If-Match": etag, "X-Request-ID": "demo-patch-0001"})
    print(f"  PATCH with If-Match {etag}   -> {ok.status_code}  new ETag {ok.headers['etag']}")
    stale = await http.patch(url, json={"note": "soy"}, headers=asha | {"If-Match": etag})
    print(f"  PATCH with stale {etag}      -> {stale.status_code}  {stale.json()['detail']}")

    heading("6. Live status over Server-Sent Events while the kitchen works")
    t0 = time.perf_counter()
    subscribed = asyncio.Event()

    async def watch() -> None:
        async with http.stream("GET", f"{url}/events", headers=asha) as stream:
            async for line in stream.aiter_lines():
                if line.startswith("data: "):
                    data = json.loads(line.removeprefix("data: "))
                    print(f"  t={time.perf_counter() - t0:4.2f}s  SSE event: status={data['status']:<10} version={data['version']}")
                    subscribed.set()
        print(f"  t={time.perf_counter() - t0:4.2f}s  stream closed by the server (order finished)")

    watcher = asyncio.create_task(watch())
    await subscribed.wait()
    etag = ok.headers["etag"]
    for status in ("paid", "preparing", "ready", "completed"):
        await asyncio.sleep(0.3)
        r = await http.post(f"{url}/status", json={"status": status}, headers=admin | {"If-Match": etag})
        etag = r.headers["etag"]
    await asyncio.wait_for(watcher, timeout=10)

    heading("7. Keyset pagination")
    for _ in range(4):
        await http.post("/v1/orders", json={"items": [{"sku": "espresso", "quantity": 1}]},
                        headers=asha | {"Idempotency-Key": secrets.token_hex(8)})
    cursor, page_no = None, 1
    while True:
        params = {"limit": 3} | ({"cursor": cursor} if cursor else {})
        page = (await http.get("/v1/orders", params=params, headers=asha)).json()
        print(f"  page {page_no}: order ids {[o['id'] for o in page['items']]}  nextCursor={page['nextCursor']}")
        cursor, page_no = page["nextCursor"], page_no + 1
        if not cursor:
            break

    heading("8. Transactional outbox -> signed webhooks (with one failure and a retry)")
    expected = query(db_file, "SELECT count(*) FROM outbox")[0][0]
    received = {}
    for _ in range(100):
        received = httpx.get(f"http://127.0.0.1:{receiver_port}/received").json()
        if received["accepted"] == expected:
            break
        await asyncio.sleep(0.2)
    for line in received["log"][:4]:
        print("  receiver:", line)
    print(f"  ... {received['accepted']} of {expected} events accepted, each signature verified")
    rows = query(db_file, "SELECT id, topic, status, attempts, last_error FROM outbox ORDER BY id LIMIT 2")
    for row in rows:
        print("  outbox row:", row)

    heading("9. Rate limiting")
    statuses = []
    for _ in range(6):
        r = await http.post("/v1/auth/token", data={"username": "ravi@example.com", "password": "guess"})
        statuses.append(r.status_code)
    print(f"  6 wrong passwords -> {statuses}  Retry-After: {r.headers.get('retry-after')} s")

    heading("10. Errors are RFC 9457 problem documents")
    r = await http.get("/v1/orders", params={"limit": 0}, headers=asha)
    print(f"  {r.status_code} {r.headers['content-type']}")
    print("  " + json.dumps(r.json(), indent=2).replace("\n", "\n  "))

    heading("11. Health checks")
    print("  GET /livez ->", (await http.get("/livez")).json(), "  GET /readyz ->", (await http.get("/readyz")).json())


if __name__ == "__main__":
    asyncio.run(main())