"""在隔离 MySQL 中验证幂等竞争、父批次串行化和连接级锁。"""

from concurrent.futures import ThreadPoolExecutor
from threading import Barrier
from uuid import uuid4

import pytest
from sqlalchemy import delete, func, select, text

from app.core.db import get_session_factory
from app.core.exceptions import BadRequestError
from app.core.tplus_lock import TPlusBusyError, sync_lock
from app.models import TPlusSyncBatch
from app.schemas.tplus import TPlusCreateIn, TPlusRetryIn
from app.services.tplus import task_service

pytestmark = pytest.mark.integration


def _parallel(action):
    barrier = Barrier(2)

    def run():
        with get_session_factory()() as session:
            barrier.wait(timeout=10)
            return action(session)

    with ThreadPoolExecutor(max_workers=2) as pool:
        futures = [pool.submit(run) for _ in range(2)]
        return [future.result(timeout=20) for future in futures]


def test_并发相同请求只创建一个批次(tplus_tasks):
    payload = TPlusCreateIn(**tplus_tasks())
    numbers = _parallel(lambda session: task_service.create_task(session, payload).batch_no)
    assert numbers[0] == numbers[1]
    with get_session_factory()() as session:
        assert (
            session.scalar(
                select(func.count())
                .select_from(TPlusSyncBatch)
                .where(TPlusSyncBatch.request_key == str(payload.request_key))
            )
            == 1
        )


def test_并发不同请求重试同父批次只允许一个(tplus_tasks):
    with get_session_factory()() as session:
        parent = task_service.create_task(session, TPlusCreateIn(**tplus_tasks()))
        parent.status = "FAILED"
        session.commit()
        number = parent.batch_no

    def retry(session):
        try:
            return task_service.retry_task(
                session, number, TPlusRetryIn(request_key=uuid4())
            ).batch_no
        except BadRequestError:
            session.rollback()
            return "REJECTED"

    try:
        numbers = _parallel(retry)
        assert numbers.count("REJECTED") == 1
        with get_session_factory()() as session:
            assert (
                session.scalar(
                    select(func.count())
                    .select_from(TPlusSyncBatch)
                    .where(TPlusSyncBatch.parent_batch_no == number)
                )
                == 1
            )
    finally:
        with get_session_factory()() as session:
            session.execute(delete(TPlusSyncBatch).where(TPlusSyncBatch.parent_batch_no == number))
            session.commit()


def test_执行锁跨提交保持且释放后可获取():
    with get_session_factory()() as owner, get_session_factory()() as other:
        with sync_lock(owner):
            owner.execute(text("SELECT 1"))
            owner.commit()
            with pytest.raises(TPlusBusyError), sync_lock(other):
                pytest.fail("另一执行者不应取得锁")
        with sync_lock(other):
            assert other.scalar(text("SELECT 1")) == 1