"""文档 12.6 第 5 条：每条改变来源明细库存的写入路径各一个并发测试。

race(): first 在事务内执行并持锁 HOLD 秒；second 在 first 持锁期间开始，
必须等 first 提交后基于最新数据判断。"""
import threading
import time

import pytest

from flyway.domain.errors import BusinessError
from flyway.models import TransferItem
from flyway.services.load_in import LoadInHeader, LoadInLine, delete_load_ins, save_load_in, set_load_in_draft
from flyway.services.load_out import (
    LoadOutHeader, LoadOutLine, delete_load_outs, save_load_out, set_load_out_draft,
)
from flyway.services.transfer import TransferHeader, TransferLine, delete_local_transfers, save_local_transfer
from tests import builders as b

HOLD = 0.6


def race(factory, first, second):
    started, results = threading.Event(), {}

    def run_first():
        try:
            with factory() as s, s.begin():
                first(s)
                started.set()
                time.sleep(HOLD)
            results["first"] = "ok"
        except Exception as e:  # noqa: BLE001
            started.set()
            results["first"] = e

    def run_second():
        started.wait()
        try:
            with factory() as s, s.begin():
                second(s)
            results["second"] = "ok"
        except Exception as e:  # noqa: BLE001
            results["second"] = e

    threads = [threading.Thread(target=run_first), threading.Thread(target=run_second)]
    for t in threads:
        t.start()
    for t in threads:
        t.join(timeout=30)
    return results


def _setup(factory, qty=100, loads=(), completed=()):
    """建客户 A、Z，来源入库明细 qty，并按 loads 建出库草稿（completed 中的序号先完成）。"""
    with factory() as s, s.begin():
        a, z = b.client(s, "A"), b.client(s, "Z")
        p = b.product(s)
        li = save_load_in(s, b.STAFF, LoadInHeader(client_id=a.id, code="LI", date=b.WHEN), [LoadInLine(p.id, "B", qty)])
        set_load_in_draft(s, b.STAFF, li.id, False)
        src = li.items[0]
        los, lines = [], []
        for n, q in enumerate(loads):
            lo = save_load_out(s, b.STAFF, LoadOutHeader(client_id=a.id, code=f"LO{n}", date=b.WHEN),
                               [LoadOutLine(src.id, q)])
            los.append(lo.id)
            lines.append(lo.items[0].id)
        for n in completed:  # 先建全部草稿再完成，否则后建的草稿会因剩余不足被拒
            set_load_out_draft(s, b.STAFF, los[n], False)
        _setup.lines = lines  # 各出库单首行 ID，供编辑类测试使用
        return a.id, z.id, li.id, src.id, los


def _remaining(factory, item_id):
    with factory() as s:
        return s.get(TransferItem, item_id).remaining_stock


def _complete(lo_id):
    return lambda s: set_load_out_draft(s, b.STAFF, lo_id, False)


def test_complete_vs_complete(factory):
    _, _, _, src, (lo1, lo2) = _setup(factory, loads=(60, 60))
    r = race(factory, _complete(lo1), _complete(lo2))
    assert r["first"] == "ok" and isinstance(r["second"], BusinessError)
    assert _remaining(factory, src) == 40


def test_revert_then_complete(factory):
    _, _, _, src, (lo1, lo2) = _setup(factory, loads=(60, 60), completed=(0,))
    r = race(factory, lambda s: set_load_out_draft(s, b.STAFF, lo1, True), _complete(lo2))
    assert r == {"first": "ok", "second": "ok"} and _remaining(factory, src) == 40


def test_delete_load_out_then_complete(factory):
    _, _, _, src, (lo1, lo2) = _setup(factory, loads=(60, 60), completed=(0,))
    r = race(factory, lambda s: delete_load_outs(s, b.ADMIN, [lo1]), _complete(lo2))
    assert r == {"first": "ok", "second": "ok"} and _remaining(factory, src) == 40


def test_completed_load_out_edit_vs_complete(factory):
    a, _, _, src, (lo1, lo2) = _setup(factory, loads=(10, 60), completed=(0,))
    line_id = _setup.lines[0]

    def edit(s):
        save_load_out(s, b.STAFF, LoadOutHeader(client_id=a, code="LO0", date=b.WHEN),
                      [LoadOutLine(src, 50, id=line_id)], transfer_id=lo1)

    r = race(factory, edit, _complete(lo2))
    assert r["first"] == "ok" and isinstance(r["second"], BusinessError)
    assert _remaining(factory, src) == 50


def test_load_in_reduce_vs_complete(factory):
    a, _, li, src, (lo1,) = _setup(factory, loads=(60,))

    def reduce(s):
        save_load_in(s, b.STAFF, LoadInHeader(client_id=a, code="LI", date=b.WHEN),
                     [LoadInLine(s.get(TransferItem, src).batch.product_id, "B", 50, id=src)], transfer_id=li)

    r = race(factory, reduce, _complete(lo1))
    assert r["first"] == "ok" and isinstance(r["second"], BusinessError)
    assert _remaining(factory, src) == 50


def test_load_in_delete_vs_load_out_save(factory):
    a, _, li, src, _ = _setup(factory)
    r = race(factory, lambda s: delete_load_ins(s, b.ADMIN, [li]),
             lambda s: save_load_out(s, b.STAFF, LoadOutHeader(client_id=a, code="LOX", date=b.WHEN),
                                     [LoadOutLine(src, 1)]))
    assert r["first"] == "ok" and isinstance(r["second"], BusinessError)


def test_transfer_vs_complete(factory):
    a, z, _, src, (lo1,) = _setup(factory, loads=(60,))
    r = race(factory,
             lambda s: save_local_transfer(s, b.STAFF, TransferHeader(a, z, "TR", b.WHEN), [TransferLine(src, 50)]),
             _complete(lo1))
    assert r["first"] == "ok" and isinstance(r["second"], BusinessError)
    assert _remaining(factory, src) == 50


def test_transfer_delete_vs_complete(factory):
    a, z, _, src, (lo1,) = _setup(factory, loads=(60,))
    with factory() as s, s.begin():
        out_id = save_local_transfer(s, b.STAFF, TransferHeader(a, z, "TR", b.WHEN), [TransferLine(src, 50)]).id
    r = race(factory, lambda s: delete_local_transfers(s, b.ADMIN, [out_id]), _complete(lo1))
    assert r == {"first": "ok", "second": "ok"} and _remaining(factory, src) == 40


def test_transfer_in_item_used_vs_transfer_reduce(factory):
    a, z, _, src, _ = _setup(factory)
    with factory() as s, s.begin():
        out = save_local_transfer(s, b.STAFF, TransferHeader(a, z, "TR", b.WHEN), [TransferLine(src, 60)])
        out_id, out_line, twin = out.id, out.items[0].id, out.transfer_tos[0].items[0].id
        lo = save_load_out(s, b.STAFF, LoadOutHeader(client_id=z, code="LOZ", date=b.WHEN), [LoadOutLine(twin, 50)])
        lo_id = lo.id
    r = race(factory, _complete(lo_id),
             lambda s: save_local_transfer(s, b.STAFF, TransferHeader(a, z, "TR", b.WHEN),
                                           [TransferLine(src, 40, id=out_line)], transfer_id=out_id))
    assert r["first"] == "ok" and isinstance(r["second"], BusinessError)
    assert _remaining(factory, twin) == 10


@pytest.mark.parametrize("reverse", [False, True])
def test_opposite_line_order_does_not_deadlock(factory, reverse):
    with factory() as s, s.begin():
        a = b.client(s)
        p = b.product(s)
        li = save_load_in(s, b.STAFF, LoadInHeader(client_id=a.id, code="LI", date=b.WHEN),
                          [LoadInLine(p.id, "X", 100), LoadInLine(p.id, "Y", 100)])
        set_load_in_draft(s, b.STAFF, li.id, False)
        x, y = (i.id for i in li.items)
        first, second = ([x, y], [y, x]) if not reverse else ([y, x], [x, y])
        lo1 = save_load_out(s, b.STAFF, LoadOutHeader(client_id=a.id, code="L1", date=b.WHEN),
                            [LoadOutLine(i, 10) for i in first]).id
        lo2 = save_load_out(s, b.STAFF, LoadOutHeader(client_id=a.id, code="L2", date=b.WHEN),
                            [LoadOutLine(i, 10) for i in second]).id
    r = race(factory, _complete(lo1), _complete(lo2))
    assert r == {"first": "ok", "second": "ok"}
    assert _remaining(factory, x) == _remaining(factory, y) == 80


def test_same_load_out_edited_twice_keeps_stored_stock_right(factory):
    with factory() as s, s.begin():
        a = b.client(s)
        p = b.product(s)
        li = save_load_in(s, b.STAFF, LoadInHeader(client_id=a.id, code="LI", date=b.WHEN),
                          [LoadInLine(p.id, n, 100) for n in ("X", "Y", "Z")])
        set_load_in_draft(s, b.STAFF, li.id, False)
        x, y, z = (i.id for i in sorted(li.items, key=lambda i: i.batch.number))
        lo = save_load_out(s, b.STAFF, LoadOutHeader(client_id=a.id, code="LO", date=b.WHEN), [LoadOutLine(x, 10)])
        set_load_out_draft(s, b.STAFF, lo.id, False)
        lo_id, line = lo.id, lo.items[0].id

    def move(target):
        return lambda s: save_load_out(s, b.STAFF, LoadOutHeader(client_id=a.id, code="LO", date=b.WHEN),
                                       [LoadOutLine(target, 10, id=line)], transfer_id=lo_id)

    r = race(factory, move(y), move(z))
    assert r == {"first": "ok", "second": "ok"}
    assert [_remaining(factory, i) for i in (x, y, z)] == [100, 100, 90]


def test_same_load_out_packed_twice_concurrently(factory):
    """两个请求同时把同一张出库单装进不同装箱单：后到的一方必须看到已装箱并被拒绝。"""
    import datetime as dt

    from flyway.models import PackingSlip, Transfer
    from flyway.services.packing import pack_load_outs
    lo_id = _setup(factory, loads=(5,), completed=(0,))[4][0]
    now = dt.datetime.now(dt.UTC)
    results = race(factory, lambda s: pack_load_outs(s, b.STAFF, [lo_id], "PS-A", now),
                   lambda s: pack_load_outs(s, b.STAFF, [lo_id], "PS-B", now))
    assert results["first"] == "ok" and isinstance(results["second"], BusinessError)
    with factory() as s:
        assert [p.reference for p in s.query(PackingSlip)] == ["PS-A"]
        assert s.get(Transfer, lo_id).packing_slip.reference == "PS-A"


def test_quick_pack_vs_slip_edit_lock_in_the_same_order(factory, monkeypatch):
    """快捷装箱与编辑装箱单同时把同一出库单放进同一装箱单：两条路径必须按同一顺序（装箱单 → 出库单）加锁，
    否则一方持出库单等装箱单、另一方持装箱单等出库单，数据库报死锁。"""
    import datetime as dt

    from sqlalchemy.exc import OperationalError

    from flyway.services import packing
    from flyway.services.packing import PackingSlipHeader, pack_load_outs, save_packing_slip
    los = _setup(factory, loads=(5, 5), completed=(0, 1))[4]
    now = dt.datetime.now(dt.UTC)
    with factory() as s, s.begin():
        slip = pack_load_outs(s, b.STAFF, [los[0]], "PS-1", now)
        slip_id, to_rmp = slip.id, slip.to_rmp
    real_lock_slip, real_load_outs = packing._lock_slip, packing._load_outs
    pack_has_first_lock, edit_has_first_lock = threading.Event(), threading.Event()

    def lock_slip(session, sid):  # 第二把锁之前停一下，让两个请求交错
        result = real_lock_slip(session, sid)
        if threading.current_thread().name == "edit":
            edit_has_first_lock.set()
            pack_has_first_lock.wait(2)
        return result

    def load_outs(session, ids):
        result = real_load_outs(session, ids)
        if threading.current_thread().name == "pack":
            pack_has_first_lock.set()
            edit_has_first_lock.wait(2)
        return result

    monkeypatch.setattr(packing, "_lock_slip", lock_slip)
    monkeypatch.setattr(packing, "_load_outs", load_outs)
    header = PackingSlipHeader(reference="PS-1", date=now, from_location="F", from_rmp="S", to_location="T",
                               to_rmp=to_rmp)
    results = {}

    def run(name, fn):
        try:
            with factory() as s, s.begin():
                fn(s)
            results[name] = "ok"
        except Exception as e:  # noqa: BLE001
            results[name] = e

    threads = [threading.Thread(name="pack", target=run, args=("pack", lambda s: pack_load_outs(s, b.STAFF, [los[1]], "PS-1", now))),
               threading.Thread(name="edit", target=run, args=("edit", lambda s: save_packing_slip(
                   s, b.STAFF, header, [los[0], los[1]], slip_id=slip_id)))]
    for t in threads:
        t.start()
    for t in threads:
        t.join(timeout=30)
    assert not any(isinstance(r, OperationalError) for r in results.values()), results
    assert sorted(type(r).__name__ if r != "ok" else "ok" for r in results.values()) == ["BusinessError", "ok"]


def test_quick_pack_rechecks_reference_after_waiting_for_the_slip_lock(factory, monkeypatch):
    """快捷装箱按编号找到装箱单后、拿到行锁前，另一人把该装箱单改了编号：取得锁后必须发现编号已变并拒绝，
    不能把出库单装进已改名的装箱单。"""
    import datetime as dt

    from flyway.models import Transfer
    from flyway.services import packing
    from flyway.services.packing import PackingSlipHeader, pack_load_outs, save_packing_slip
    los = _setup(factory, loads=(5, 5), completed=(0, 1))[4]
    now = dt.datetime.now(dt.UTC)
    with factory() as s, s.begin():
        slip = pack_load_outs(s, b.STAFF, [los[0]], "PS-1", now)
        slip_id, to_rmp = slip.id, slip.to_rmp
    real_lock_slip = packing._lock_slip
    found, renamed = threading.Event(), threading.Event()

    def lock_slip(session, sid):
        if threading.current_thread().name == "pack":
            found.set()       # 已按编号找到装箱单，尚未加锁
            renamed.wait(5)   # 等另一人改完编号并提交
        return real_lock_slip(session, sid)

    monkeypatch.setattr(packing, "_lock_slip", lock_slip)
    results = {}

    def pack():
        try:
            with factory() as s, s.begin():
                pack_load_outs(s, b.STAFF, [los[1]], "PS-1", now)
            results["pack"] = "ok"
        except Exception as e:  # noqa: BLE001
            results["pack"] = e

    t = threading.Thread(name="pack", target=pack)
    t.start()
    found.wait(5)
    with factory() as s, s.begin():
        header = PackingSlipHeader(reference="PS-RENAMED", date=now, from_location="F", from_rmp="S", to_location="T",
                                   to_rmp=to_rmp)
        save_packing_slip(s, b.STAFF, header, [los[0]], slip_id=slip_id)
    renamed.set()
    t.join(timeout=30)
    assert isinstance(results["pack"], BusinessError), results
    with factory() as s:
        assert s.get(Transfer, los[1]).packing_slip_id is None