import sys
import time
from concurrent.futures import ThreadPoolExecutor

# Strict dependency resolution and defensive import verification
try:
    import orjson
except ModuleNotFoundError:
    print("Error: 'orjson' is not installed. Please run 'pip install -r requirements.txt'.", file=sys.stderr)
    sys.exit(1)

try:
    from pydantic import BaseModel, Field
except ModuleNotFoundError:
    print("Error: 'pydantic' is not installed. Please run 'pip install -r requirements.txt'.", file=sys.stderr)
    sys.exit(1)

try:
    from sqlalchemy import create_engine, Integer, String, Text, select
    from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker
    from sqlalchemy.exc import OperationalError
except ModuleNotFoundError:
    print("Error: 'sqlalchemy' is not installed. Please run 'pip install -r requirements.txt'.", file=sys.stderr)
    sys.exit(1)


def serialize_result(model_dump: dict) -> str:
    """High-throughput deterministic serialization via orjson."""
    return orjson.dumps(model_dump).decode("utf-8")


# ==========================================
# 1. Pydantic v2 Schema Definitions & Validation
# ==========================================
class AccountModel(BaseModel):
    account_id: int = Field(..., description="Unique Account Identifier")
    balance: int = Field(..., description="Account balance in fractional units")
    memo: str = Field(..., description="Audit trace or mutation context")

    class Config:
        frozen = True


class ExecutionResultModel(BaseModel):
    thread_name: str
    success: bool
    message: str
    elapsed_time: float

    class Config:
        frozen = True


# ==========================================
# 2. SQLAlchemy 2.0 Declarative ORM Model
# ==========================================
class Base(DeclarativeBase):
    pass


class Account(Base):
    __tablename__ = "accounts"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=False)
    balance: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
    memo: Mapped[str] = mapped_column(String(255), nullable=True)


# ==========================================
# 3. Deadlock Simulation & Concurrency Harness
# ==========================================
def initialize_database(engine):
    """Purge and provision test tables with initial ledger balances."""
    Base.metadata.drop_all(bind=engine)
    Base.metadata.create_all(bind=engine)

    SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)
    with SessionLocal() as session:
        with session.begin():
            session.add(Account(id=1, balance=1000, memo="Initial"))
            session.add(Account(id=2, balance=1000, memo="Initial"))


def worker_transaction(
    db_url: str, 
    thread_name: str, 
    source_id: int, 
    target_id: int, 
    amount: int, 
    autoflush: bool
) -> str:
    """
    Concurrent worker transaction routine.
    Simulates statement inversion caused by implicit autoflush triggering race conditions,
    or validates deterministic execution using autoflush=False and explicit transaction boundaries.
    """
    engine = create_engine(
        db_url, 
        isolation_level="SERIALIZABLE",  # Enforce strict serialization to highlight concurrency conflicts
        future=True
    )
    SessionLocal = sessionmaker(bind=engine, autoflush=autoflush, autocommit=False)

    start_time = time.perf_counter()
    success = False
    msg = ""

    try:
        with SessionLocal() as session:
            # 1. State Hydration (SELECT phase)
            source_acc = session.get(Account, source_id)
            target_acc = session.get(Account, target_id)

            if not source_acc or not target_acc:
                raise ValueError("Account not found")

            # 2. Domain Logic Execution (Validated through Pydantic models)
            validated_source = AccountModel(
                account_id=source_acc.id,
                balance=source_acc.balance - amount,
                memo=f"Transferred to {target_id}"
            )
            validated_target = AccountModel(
                account_id=target_acc.id,
                balance=target_acc.balance + amount,
                memo=f"Received from {source_id}"
            )

            # 3. In-Memory ORM State Mutation
            source_acc.balance = validated_source.balance
            source_acc.memo = validated_source.memo

            # [The Autoflush Hazard]
            # When autoflush=True, any subsequent query or internal session dependency triggers an
            # implicit flush, dispatching intermediate DML statements before the developer intended.
            target_acc.balance = validated_target.balance
            target_acc.memo = validated_target.memo

            # Explicit Commit Boundary
            session.commit()
            success = True
            msg = "Transaction committed successfully."
    except OperationalError as oe:
        msg = f"OperationalError (Deadlock/Conflict): {oe.orig}"
    except Exception as e:
        msg = f"Error: {str(e)}"
    finally:
        engine.dispose()

    elapsed = time.perf_counter() - start_time

    # Model serialization via orjson
    result_model = ExecutionResultModel(
        thread_name=thread_name,
        success=success,
        message=msg,
        elapsed_time=elapsed
    )
    return serialize_result(result_model.model_dump())


def main():
    db_url = "sqlite:///./deadlock_test.db"

    engine = create_engine(db_url, future=True)
    initialize_database(engine)
    engine.dispose()

    print("--- Case 1: Autoflush = True (Premature Flush & Uncontrolled Order Risk) ---")
    with ThreadPoolExecutor(max_workers=2) as executor:
        f1 = executor.submit(worker_transaction, db_url, "Thread-A", 1, 2, 100, True)
        f2 = executor.submit(worker_transaction, db_url, "Thread-B", 2, 1, 50, True)
        print(f"Result A: {f1.result()}")
        print(f"Result B: {f2.result()}")

    initialize_database(create_engine(db_url, future=True))

    print("\n--- Case 2: Autoflush = False & Explicit Control (Safe Locking Boundary) ---")
    with ThreadPoolExecutor(max_workers=2) as executor:
        f3 = executor.submit(worker_transaction, db_url, "Thread-A", 1, 2, 100, False)
        f4 = executor.submit(worker_transaction, db_url, "Thread-B", 2, 1, 50, False)
        print(f"Result A: {f3.result()}")
        print(f"Result B: {f4.result()}")


if __name__ == "__main__":
    main()

# Acquire locks in strictly sorted primary key order to eliminate cyclic waits
ordered_ids = sorted([source_id, target_id])
stmt = (
    select(Account)
    .where(Account.id.in_(ordered_ids))
    .with_for_update()
    .order_by(Account.id)
)
locked_accounts = session.scalars(stmt).all()