"""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 http.cookiejar
import json
import secrets
import struct
import zlib
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
from contextlib import contextmanager

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):
    return call(urllib.request.urlopen, BASE, path, method, body)


def call(opener, base, 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(opener(r, timeout=60))


def make_png(w=40, h=40, rgb=(120, 60, 200)):
    raw = b"".join(b"\x00" + bytes(rgb) * w for _ in range(h))

    def chunk(tag, data):
        body = tag + data
        return struct.pack(">I", len(data)) + body + struct.pack(">I", zlib.crc32(body))

    return (b"\x89PNG\r\n\x1a\n"
            + chunk(b"IHDR", struct.pack(">IIBBBBB", w, h, 8, 2, 0, 0, 0))
            + chunk(b"IDAT", zlib.compress(raw))
            + chunk(b"IEND", b""))


def multipart(content, filename="photo.png", ctype="image/png", fields=None):
    """Build a multipart body by hand — no third-party HTTP client in the tests."""
    boundary = "----parts" + secrets.token_hex(8)
    out = []
    for key, value in (fields or {}).items():
        out.append(f"--{boundary}\r\nContent-Disposition: form-data; "
                   f'name="{key}"\r\n\r\n{value}\r\n'.encode())
    out.append(f"--{boundary}\r\nContent-Disposition: form-data; "
               f'name="file"; filename="{filename}"\r\n'
               f"Content-Type: {ctype}\r\n\r\n".encode() + content + b"\r\n")
    out.append(f"--{boundary}--\r\n".encode())
    return b"".join(out), f"multipart/form-data; boundary={boundary}"


def post_image(base, part_id, content, timeout=30):
    body, ctype = multipart(content)
    r = urllib.request.Request(f"{base}/api/parts/{part_id}/images", data=body,
                               method="POST", headers={"Content-Type": ctype})
    try:
        with urllib.request.urlopen(r, timeout=timeout) as resp:
            return resp.status
    except urllib.error.HTTPError as e:
        return e.code


@contextmanager
def server(port, **env_extra):
    """Run a real uvicorn against a throwaway database."""
    tmp = tempfile.mkdtemp()
    env = {**os.environ, "PARTS_DB": os.path.join(tmp, "conc.db"), **env_extra}
    proc = 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,
    )
    base = f"http://127.0.0.1:{port}"
    try:
        for _ in range(120):
            try:
                call(urllib.request.urlopen, base, "/healthz")
                break
            except Exception:
                time.sleep(0.25)
        else:
            raise RuntimeError("server never started")
        yield base, tmp
    finally:
        proc.send_signal(signal.SIGINT)
        try:
            proc.wait(timeout=10)
        except subprocess.TimeoutExpired:
            proc.kill()
        shutil.rmtree(tmp, ignore_errors=True)


def session_for(base, password):
    """Log in and return an opener holding that session's cookie."""
    jar = http.cookiejar.CookieJar()
    opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar))
    call(opener.open, base, "/api/login", "POST", {"password": password})
    return opener


def authenticated(opener, base) -> bool:
    try:
        call(opener.open, base, "/api/parts")
        return True
    except urllib.error.HTTPError as e:
        if e.code == 401:
            return False
        raise


def session_phase():
    """Concurrent session revocation, which needs the auth gate switched on."""
    password = "concurrency-test-password"
    with server(PORT + 1, PARTS_AUTH="on", PARTS_PASSWORD=password,
                PARTS_SECRET="concurrency-test-secret",
                PARTS_LOGIN_MAX_FAILURES="500") as (base, _tmp):
        # --- the plain case first ---
        keeper = session_for(base, password)
        others = [session_for(base, password) for _ in range(5)]
        check("sessions start authenticated",
              all(authenticated(o, base) for o in [keeper] + others))
        call(keeper.open, base, "/api/sessions/revoke", "POST", {})
        check("revoke keeps the calling session", authenticated(keeper, base))
        check("revoke signs out every other session",
              not any(authenticated(o, base) for o in others))

        # --- concurrent revocation ---
        # Each successful revoke bumps the epoch and re-issues a cookie carrying
        # the value it wrote, so requests already in flight legitimately come
        # back 401 as earlier ones invalidate them. Exactly one session — the
        # one that wrote the final epoch — should be left standing.
        #
        # That is precisely what a lost increment breaks: several bumps read the
        # same epoch, all write the same value, all re-issue cookies matching
        # it, and most of the sessions that were meant to be cut off survive.
        openers = [session_for(base, password) for _ in range(20)]
        check("all 20 sessions start authenticated",
              all(authenticated(o, base) for o in openers))

        def try_revoke(opener):
            try:
                call(opener.open, base, "/api/sessions/revoke", "POST", {})
                return 200
            except urllib.error.HTTPError as e:
                return e.code

        with ThreadPoolExecutor(max_workers=20) as ex:
            codes = list(ex.map(try_revoke, openers))
        check("at least one concurrent revoke succeeded", 200 in codes, str(codes))
        check("losers were rejected rather than erroring",
              all(c in (200, 401) for c in codes), str(sorted(set(codes))))
        alive = [o for o in openers if authenticated(o, base)]
        check("concurrent revokes leave exactly one session standing",
              len(alive) == 1, f"{len(alive)} of 20 still authenticated")
        if alive:
            check("the surviving session still works",
                  call(alive[0].open, base, "/api/stats")["parts"] == 0)
        fresh = session_for(base, password)
        check("a new login works after mass revocation", authenticated(fresh, base))


def main():
    with server(PORT, PARTS_AUTH="off") as (_base, tmp):
        # --- 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}")

        # --- taxonomy renames must leave the index agreeing with the tree ---
        # The scan for affected parts and the reindex that follows have to see
        # one consistent tree. Without the write lock a concurrent rename slips
        # between them, and search keeps matching a name the UI no longer shows.
        # Names are chosen so no token is a prefix of another: search uses prefix
        # matching, so "Taxo 1" would legitimately match "Taxo 19" and the test
        # would report a race that isn't there.
        WORDS = ["alfa", "bravo", "charlie", "delta", "echo", "foxtrot", "golf",
                 "hotel", "india", "juliett", "kilo", "lima", "mike", "november",
                 "oscar", "papa", "quebec", "romeo", "sierra", "tango"]
        cat = req("/api/categories", "POST", {"name": "Taxo zulu"})["id"]
        req("/api/parts", "POST", {"name": "taxo widget", "category_id": cat, "quantity": 1})
        names = [f"Taxo {w}" for w in WORDS]
        with ThreadPoolExecutor(max_workers=20) as ex:
            list(ex.map(lambda n: req(f"/api/categories/{cat}", "PATCH", {"name": n}), names))
        final = [c["name"] for c in req("/api/categories")["items"] if c["id"] == cat][0]
        hits_final = req(f"/api/parts?q={final.replace(' ', '+')}")["total"]
        check("search matches the category's final name", hits_final == 1, f"{final} -> {hits_final}")
        stale = [n for n in ["Taxo zulu"] + names if n != final
                 and req(f"/api/parts?q={n.replace(' ', '+')}")["total"] > 0]
        check("no superseded category name still matches", not stale, f"stale: {stale}")

        # Same race with a location, and with parts being created concurrently.
        loc = req("/api/locations", "POST", {"name": "Loc zulu"})["id"]
        loc_names = [f"Loc {w}" for w in WORDS[:15]]

        def rename_or_add(i):
            if i % 3 == 0:
                req("/api/parts", "POST", {"name": f"loc widget {i}", "location_id": loc, "quantity": 1})
            else:
                req(f"/api/locations/{loc}", "PATCH", {"name": loc_names[i % len(loc_names)]})

        with ThreadPoolExecutor(max_workers=15) as ex:
            list(ex.map(rename_or_add, range(15)))
        final_loc = [l["name"] for l in req("/api/locations")["items"] if l["id"] == loc][0]
        in_loc = req(f"/api/parts?location_id={loc}&limit=100")["total"]
        by_name = req(f"/api/parts?q={final_loc.replace(' ', '+')}&limit=100")["total"]
        check("every part in the location is indexed under its final name",
              by_name == in_loc, f"filter={in_loc} search={by_name}")
        stale_loc = [n for n in ["Loc zulu"] + loc_names if n != final_loc
                     and req(f"/api/parts?q={n.replace(' ', '+')}&limit=100")["total"] > 0]
        check("no superseded location name still matches", not stale_loc, f"stale: {stale_loc}")

        # --- 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"]))
        # --- concurrent photo uploads ---
        # An async endpoint doing blocking SQLite work stalls the whole event
        # loop under contention; an unlocked count lets every overlapping upload
        # observe the same total and sail past the ceiling together. This checks
        # both: the ceiling holds exactly, and the server stays responsive while
        # it is being hammered.
        shot_part = req("/api/parts", "POST", {"name": "photo race", "quantity": 1})["id"]
        attempts = 20
        latency = []

        def ping():
            start = time.time()
            try:
                call(urllib.request.urlopen, _base, "/healthz")
            except Exception:
                pass
            latency.append(time.time() - start)

        with ThreadPoolExecutor(max_workers=attempts + 4) as ex:
            futures = [ex.submit(post_image, _base, shot_part, make_png(60, 60, (i * 9, 40, 90)))
                       for i in range(attempts)]
            for _ in range(4):
                ex.submit(ping)
            codes = [f.result() for f in futures]

        created = codes.count(201)
        refused = codes.count(409)
        check("no upload returned a server error",
              not [c for c in codes if c >= 500], str(sorted(set(codes))))
        check("every upload was answered", len(codes) == attempts and None not in codes)
        check("exactly the ceiling was stored", created == 12, f"{created} created, {refused} refused")
        check("the rest were refused as over the limit", created + refused == attempts,
              str(sorted(set(codes))))
        stored = req(f"/api/parts/{shot_part}/images")["items"]
        check("the database agrees with the ceiling", len(stored) == 12, str(len(stored)))
        check("positions are unique", len({i["position"] for i in stored}) == 12,
              str(sorted(i["position"] for i in stored)))

        image_dir = os.path.join(tmp, "images")
        on_disk = len(os.listdir(image_dir))
        check("refused uploads left no files behind", on_disk == 12, f"{on_disk} files")

        worst = max(latency) if latency else 0
        # A blocked event loop parks unrelated requests for SQLite's busy
        # timeout, which is 15s. Anything in that neighbourhood means the
        # upload is running on the loop again.
        check("the server stayed responsive during the upload storm",
              worst < 5.0, f"slowest /healthz was {worst:.1f}s")

    session_phase()

    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())