"""Phase 5 shop writes under concurrency: weld and heat tag numbers come from the packet counter under the row
lock, so two sessions marking at once never collide; two supervisors verify two departments of one change order
at once and both land; two users tick different checklist keys and both stay; a stale expect is refused with the
message; a write to a packet filed in the meantime is refused."""

from __future__ import annotations

import threading
from collections.abc import Callable, Iterator
from typing import Any

import pytest
from fakedir import FakeDirectory
from fastapi.testclient import TestClient
from pkthelpers import step
from pshelpers import needs_pg
from shophelpers import (
    SIGN,
    STALE,
    Shop,
    all_rows,
    finish_stage,
    fresh,
    in_progress,
    issue_co,
    ok,
    open_shop,
    refused,
    run,
    to_station,
    verify,
)

from packetstation.settings import Settings

pytestmark = needs_pg


@pytest.fixture
def shop(pg_settings: Settings, fake_dir: FakeDirectory, dev_visual: None) -> Iterator[Shop]:
    yield from open_shop(pg_settings, fake_dir)


def together(calls: list[Callable[[], Any]]) -> list[Any]:
    """Run the calls in threads released by one barrier. Returns their answers in order."""
    results: list[Any] = [None] * len(calls)
    errors: list[BaseException] = []
    barrier = threading.Barrier(len(calls))

    def run_one(i: int) -> None:
        try:
            barrier.wait(10)
            results[i] = calls[i]()
        except BaseException as e:  # reported below
            errors.append(e)

    threads = [threading.Thread(target=run_one, args=(i,)) for i in range(len(calls))]
    for t in threads:
        t.start()
    for t in threads:
        t.join(60)
    assert not errors, errors
    return results


def weld_call(c: TestClient, pid: int, fx: float) -> Callable[[], Any]:
    body = {"page": 1, "fx": fx, "fy": 0.5, "items": [1], "type": "Fillet", "process": "GMAW"}
    return lambda: c.post(f"/api/packets/{pid}/welds", json=body)


def tag_call(c: TestClient, pid: int, fx: float) -> Callable[[], Any]:
    return lambda: c.post(f"/api/packets/{pid}/heat-tags", json={"page": 1, "item": 2, "fx": fx, "fy": 0.4})


def test_concurrent_welds_never_collide(shop: Shop) -> None:
    pkt = in_progress(shop)
    pid = pkt["id"]
    # weld.andy on two sessions (two tablets) and sup.wes, three welds each, all at once
    clients = [shop.fresh("weld.andy"), shop.fresh("weld.andy"), shop.fresh("sup.wes")]
    calls = [weld_call(c, pid, 0.1 + 0.1 * i + 0.01 * k) for i, c in enumerate(clients) for k in range(3)]
    answers = together(calls)
    assert [r.status_code for r in answers] == [200] * len(calls), [r.text for r in answers if r.status_code != 200]
    nos = sorted(w["weld_no"] for w in fresh(shop("sup.wes"), pid)["welds"])
    assert nos == list(range(1, len(calls) + 1))
    rows = all_rows(shop.settings, "SELECT weld_no FROM welds WHERE packet_id = %s ORDER BY weld_no", (pid,))
    assert [r[0] for r in rows] == nos
    marked = sorted(
        r[0]
        for r in all_rows(shop.settings, "SELECT action FROM packet_log WHERE packet_id = %s", (pid,))
        if r[0].endswith(" marked") and r[0].startswith("Weld W")
    )
    assert marked == sorted(f"Weld W{n} marked" for n in nos)
    assert (
        int(all_rows(shop.settings, "SELECT next_weld_no FROM packets WHERE id = %s", (pid,))[0][0]) == len(calls) + 1
    )


def test_two_welds_at_once_are_w1_and_w2(shop: Shop) -> None:
    pkt = in_progress(shop)
    a, b = together(
        [weld_call(shop.fresh("weld.andy"), pkt["id"], 0.2), weld_call(shop.fresh("sup.wes"), pkt["id"], 0.7)]
    )
    assert a.status_code == 200 and b.status_code == 200, (a.text, b.text)
    got = {w["fx"]: w["weld_no"] for w in fresh(shop("sup.wes"), pkt["id"])["welds"]}
    assert sorted(got.values()) == [1, 2] and got[0.2] != got[0.7]


def test_concurrent_heat_tags_never_collide(shop: Shop) -> None:
    pkt = in_progress(shop)
    pid = pkt["id"]
    clients = [shop.fresh("weld.andy"), shop.fresh("sup.wes"), shop.fresh("lead.lou")]
    calls = [tag_call(c, pid, 0.1 + 0.1 * i + 0.01 * k) for i, c in enumerate(clients) for k in range(3)]
    answers = together(calls)
    assert [r.status_code for r in answers] == [200] * len(calls), [r.text for r in answers if r.status_code != 200]
    nos = sorted(t["tag_no"] for t in fresh(shop("sup.wes"), pid)["heat_tags"])
    assert nos == list(range(1, len(calls) + 1))


def test_two_supervisors_verify_at_once(shop: Shop) -> None:
    pkt = in_progress(shop)
    pkt = issue_co(shop, pkt, ["03", "06"])
    assert pkt["change_orders"][0]["required"] == ["03", "06"]
    wes, mia = shop.fresh("sup.wes"), shop.fresh("multi.mia")

    def call(c: TestClient, dept: str) -> Callable[[], Any]:
        body = {"dept": dept, "note": f"Checked in {dept}", "sign": SIGN}
        return lambda: c.post(f"/api/packets/{pkt['id']}/change-orders/1/verify", json=body)

    a, b = together([call(wes, "03"), call(mia, "06")])
    assert a.status_code == 200 and b.status_code == 200, (a.text, b.text)
    out = fresh(shop("sup.wes"), pkt["id"])
    co = out["change_orders"][0]
    assert set(co["verified"]) == {"03", "06"} and co["pending_depts"] == [] and co["open"] is False
    assert out["open_cos"] == []
    rows = all_rows(shop.settings, "SELECT dept_code FROM co_verifications ORDER BY dept_code")
    assert [r[0] for r in rows] == ["03", "06"]
    notes = sorted(
        r[0]
        for r in all_rows(
            shop.settings,
            "SELECT note FROM packet_log WHERE packet_id = %s AND action LIKE 'Change order CO-1 verified by %%'",
            (pkt["id"],),
        )
    )
    # one of them landed first and still waited on the other; the second fully verified it
    assert len(notes) == 2
    assert any(n.endswith("change order fully verified") for n in notes)
    assert any("still waiting on" in n for n in notes)
    # the same department twice at once: one lands, one is told it is done
    pkt2 = issue_co(shop, out, ["03"])
    no = pkt2["change_orders"][-1]["co_no"]

    def again(c: TestClient) -> Callable[[], Any]:
        return lambda: c.post(
            f"/api/packets/{pkt['id']}/change-orders/{no}/verify", json={"dept": "03", "note": "x", "sign": SIGN}
        )

    a, b = together([again(shop.fresh("sup.wes")), again(shop.fresh("sup.wes"))])
    assert sorted([a.status_code, b.status_code]) == [200, 409], (a.text, b.text)
    loser = a if a.status_code == 409 else b
    assert loser.json()["code"] == "already_verified"


def test_checklist_keys_merge(shop: Shop) -> None:
    pkt = in_progress(shop)
    andy, wes = shop.fresh("weld.andy"), shop.fresh("sup.wes")
    url = f"/api/packets/{pkt['id']}/stage/checklist"
    pairs = [(andy, "dwg"), (wes, "heat"), (andy, "wps"), (wes, "fit")]
    answers = together([(lambda c=c, k=k: c.put(f"{url}/{k}", json={"done": True})) for c, k in pairs])
    assert [r.status_code for r in answers] == [200] * 4, [r.text for r in answers]
    ck = fresh(shop("sup.wes"), pkt["id"])["stage"]["checklist"]
    assert {k for k, v in ck.items() if v.get("done")} == {"dwg", "heat", "wps", "fit"}
    assert ck["dwg"]["by"] == shop.uid("weld.andy") and ck["heat"]["by"] == shop.uid("sup.wes")


def test_stale_expect_is_refused(shop: Shop) -> None:
    pkt = finish_stage(shop, in_progress(shop))
    andy_view = fresh(shop("weld.andy"), pkt["id"])
    ok(step(shop("sup.wes"), pkt, "pull_back"))
    r = refused(step(shop("weld.andy"), andy_view, "submit", sign=SIGN), 409, "stale_step", STALE)
    assert r["current"]["status"] == "ready" and r["current"]["dept"] == "03"
    assert r["current"]["version"] == fresh(shop("sup.wes"), pkt["id"])["version"]
    # pull back and start at once: one of them is stale
    pkt = to_station(shop, fresh(shop("sup.wes"), pkt["id"]))
    last = all_rows(
        shop.settings, "SELECT coalesce(max(id), 0) FROM packet_transitions WHERE packet_id = %s", (pkt["id"],)
    )
    wes, andy = shop.fresh("sup.wes"), shop.fresh("weld.andy")
    a, b = together([lambda: step(wes, pkt, "unassign"), lambda: step(andy, pkt, "start")])
    assert sorted([a.status_code, b.status_code]) == [200, 409], (a.text, b.text)
    loser = a if a.status_code == 409 else b
    assert loser.json()["code"] == "stale_step" and loser.json()["detail"] == STALE
    n = all_rows(
        shop.settings,
        "SELECT count(*) FROM packet_transitions WHERE packet_id = %s AND id > %s",
        (pkt["id"], last[0][0]),
    )[0][0]
    assert n == 1


def test_filed_in_the_meantime_is_refused(shop: Shop) -> None:
    pkt = in_progress(shop)
    pkt = issue_co(shop, pkt, ["03"])
    seen = fresh(shop("sup.wes"), pkt["id"])
    run(shop.settings, "UPDATE packets SET status = 'filed', version = version + 1 WHERE id = %s", (pkt["id"],))
    msg = "This packet is filed and locked. IT can reopen it if a correction is needed."
    refused(verify(shop, seen, 1, "03", "sup.wes"), 403, "filed", msg)
    refused(
        shop("clerk.clara").patch(f"/api/packets/{pkt['id']}", json={"set": {"drawing": "D-9"}, "base": {}}),
        403,
        "filed",
        msg,
    )
    refused(verify(shop, seen, 1, "03", "admin.ann", it_reason="Ticket 5502: verify after filing"), 403, "filed", msg)
    # a shop edit is not open on a filed packet either
    r = shop("weld.andy").post(
        f"/api/packets/{pkt['id']}/welds", json={"page": 1, "fx": 0.5, "fy": 0.5, "items": [1], "type": "Fillet"}
    )
    assert r.status_code == 403 and r.json()["code"] in ("filed", "wrong_status"), r.text
    r = step(shop("weld.andy"), seen, "submit", sign=SIGN)
    assert r.status_code == 409 and r.json()["code"] == "stale_step", r.text