"""Concurrency checks against a real uvicorn process.

The in-process test client is not enough here: a lost update needs two requests
genuinely overlapping inside SQLite, which means a real server and real threads.
"""

import json
import os
import shutil
import signal
import subprocess
import sys
import tempfile
import time
import urllib.error
import urllib.request
from concurrent.futures import ThreadPoolExecutor

PORT = 8137
BASE = f"http://127.0.0.1:{PORT}"
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))

failures = []


def check(label, condition, detail=""):
    print(("  PASS  " if condition else "  FAIL  ") + label + (f"  [{detail}]" if detail and not condition else ""))
    if not condition:
        failures.append(label)


def req(path, method="GET", body=None):
    data = json.dumps(body).encode() if body is not None else None
    r = urllib.request.Request(BASE + path, data=data, method=method,
                               headers={"Content-Type": "application/json"})
    return json.load(urllib.request.urlopen(r, timeout=60))


def main():
    tmp = tempfile.mkdtemp()
    env = {
        **os.environ,
        "PARTS_DB": os.path.join(tmp, "conc.db"),
        "PARTS_AUTH": "off",
    }
    server = subprocess.Popen(
        [sys.executable, "-m", "uvicorn", "app.main:app", "--port", str(PORT), "--log-level", "warning"],
        cwd=ROOT, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
    )
    try:
        for _ in range(120):
            try:
                req("/healthz")
                break
            except Exception:
                time.sleep(0.25)
        else:
            raise RuntimeError("server never started")

        # --- concurrent decrements must not lose updates ---
        pid = req("/api/parts", "POST", {"name": "race target", "quantity": 100})["id"]
        with ThreadPoolExecutor(max_workers=50) as ex:
            codes = list(ex.map(
                lambda _: req(f"/api/parts/{pid}/adjust", "POST", {"delta": -1, "reason": "race"}) and 200,
                range(50)))
        check("all 50 concurrent adjustments succeeded", codes.count(200) == 50)
        final = req(f"/api/parts/{pid}")["quantity"]
        check("50 concurrent -1 adjustments land at 50", final == 50, f"got {final}")
        hist = req(f"/api/parts/{pid}/history?limit=500")["items"]
        check("history has one row per change", len(hist) == 51, f"got {len(hist)}")
        total = sum(h["delta"] for h in hist)
        check("stock log sums to the stored quantity", total == final, f"log={total} stored={final}")

        # --- mixed concurrent increments and decrements ---
        # Start high enough that no intermediate ordering can hit the floor at
        # zero: with the floor in play the net is legitimately order-dependent,
        # which would make this assertion about clamping rather than atomicity.
        start = 100.0
        deltas = [5] * 30 + [-3] * 30
        pid2 = req("/api/parts", "POST", {"name": "mixed target", "quantity": start})["id"]
        with ThreadPoolExecutor(max_workers=30) as ex:
            list(ex.map(lambda d: req(f"/api/parts/{pid2}/adjust", "POST", {"delta": d}), deltas))
        got = req(f"/api/parts/{pid2}")["quantity"]
        check("mixed concurrent adjustments net out correctly",
              got == start + sum(deltas), f"got {got}, expected {start + sum(deltas)}")
        h2 = req(f"/api/parts/{pid2}/history?limit=500")["items"]
        check("mixed adjustment log sums to the stored quantity",
              sum(x["delta"] for x in h2) == got, f"log={sum(x['delta'] for x in h2)} stored={got}")

        # --- the floor at zero is still honoured under contention ---
        pid_floor = req("/api/parts", "POST", {"name": "floor race", "quantity": 10})["id"]
        with ThreadPoolExecutor(max_workers=20) as ex:
            list(ex.map(lambda _: req(f"/api/parts/{pid_floor}/adjust", "POST", {"delta": -1}), range(20)))
        fq = req(f"/api/parts/{pid_floor}")["quantity"]
        fh = req(f"/api/parts/{pid_floor}/history?limit=500")["items"]
        check("20 concurrent -1 on a stock of 10 floors at zero", fq == 0, f"got {fq}")
        check("floored concurrent log still sums to the stored quantity",
              sum(x["delta"] for x in fh) == fq, f"log={sum(x['delta'] for x in fh)} stored={fq}")

        # --- concurrent PATCH quantity is logged exactly once per change ---
        pid3 = req("/api/parts", "POST", {"name": "patch target", "quantity": 0})["id"]
        with ThreadPoolExecutor(max_workers=20) as ex:
            list(ex.map(lambda i: req(f"/api/parts/{pid3}", "PATCH", {"quantity": float(i + 1)}), range(20)))
        h3 = req(f"/api/parts/{pid3}/history?limit=500")["items"]
        stored = req(f"/api/parts/{pid3}")["quantity"]
        check("concurrent PATCHes each logged a row", len(h3) == 20, f"got {len(h3)}")
        check("PATCH log's final quantity_after matches stored",
              h3[0]["quantity_after"] == stored, f"log={h3[0]['quantity_after']} stored={stored}")

        # --- concurrent creates don't collide ---
        with ThreadPoolExecutor(max_workers=25) as ex:
            ids = list(ex.map(
                lambda i: req("/api/parts", "POST", {"name": f"bulk {i}", "quantity": 1})["id"], range(25)))
        check("25 concurrent creates produced 25 distinct parts", len(set(ids)) == 25)
        check("all concurrent creates are searchable",
              req("/api/parts?q=bulk&limit=100")["total"] == 25,
              str(req("/api/parts?q=bulk&limit=100")["total"]))
    finally:
        server.send_signal(signal.SIGINT)
        try:
            server.wait(timeout=10)
        except subprocess.TimeoutExpired:
            server.kill()
        shutil.rmtree(tmp, ignore_errors=True)

    print()
    if failures:
        print(f"{len(failures)} FAILED: " + "; ".join(failures))
        return 1
    print("all concurrency checks passed")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())