"""Какие аномалии уровень изоляции пропускает — воспроизведением, а не по таблице.

ЗАЧЕМ ЭТОТ СКРИПТ. Про уровни изоляции почти всегда рассказывают таблицей из
стандарта SQL: четыре уровня, три аномалии, галочки. Таблица описывает, что
уровню РАЗРЕШЕНО пропускать, а не что он пропускает у вас. Разница
принципиальная в обе стороны, и обе видны только запуском:

  * PostgreSQL СТРОЖЕ таблицы там, где стандарт разрешает быть слабее.
    READ UNCOMMITTED принимается — `transaction_isolation` возвращает именно
    его, — но ведёт себя как READ COMMITTED: незафиксированное не читается,
    и грязное чтение недостижимо ни на одном уровне. А REPEATABLE READ
    реализован снимком, поэтому фантомы, которые стандарт на этом уровне
    разрешает, не воспроизводятся.

  * И при этом СЛАБЕЕ ожиданий там, где таблицы вообще молчат. Write skew —
    аномалия, которой в списке стандарта нет, — на REPEATABLE READ проходит
    целиком. Два перевода, каждый по отдельности законный, вместе нарушают
    инвариант, ради которого их и писали.

Скрипт воспроизводит четыре аномалии на трёх уровнях и печатает, что
случилось на самом деле. Никаких пауз-угадаек: порядок шагов задан явно двумя
соединениями, поэтому результат не зависит от скорости машины.

ЗАПУСК. Нужен живой PostgreSQL; адрес берётся из DE_BENCH_DSN, иначе
локальный по умолчанию:

    DE_BENCH_DSN="postgres://postgres@localhost:5433/postgres" \\
      python3 bench/isolation/anomalies.py
"""

import os
import sys

import psycopg

DSN = os.environ.get("DE_BENCH_DSN", "postgres://postgres@localhost:5433/postgres")

LEVELS = ["READ COMMITTED", "REPEATABLE READ", "SERIALIZABLE"]

# В psycopg 3 классы ошибок разложены по SQLSTATE ПЛОСКО: SerializationFailure
# и DeadlockDetected наследуются напрямую от OperationalError, а не от общего
# TransactionRollback (он существует, но их предком не является). Ловить класс
# 40 одним именем нельзя — только перечислением.
ROLLBACK = (psycopg.errors.SerializationFailure, psycopg.errors.DeadlockDetected)


def fresh(conn, sql: str) -> None:
    with conn.cursor() as cur:
        cur.execute(sql)


def setup() -> None:
    """Две таблицы: счёт для чтений и дежурства для write skew."""
    with psycopg.connect(DSN, autocommit=True) as conn:
        fresh(conn, "DROP TABLE IF EXISTS accounts, oncall")
        fresh(conn, "CREATE TABLE accounts (id int primary key, balance int not null)")
        fresh(conn, "INSERT INTO accounts VALUES (1, 100), (2, 100)")
        fresh(conn, "CREATE TABLE oncall (doctor text primary key, on_call bool not null)")
        fresh(conn, "INSERT INTO oncall VALUES ('Алиса', true), ('Борис', true)")


def begin(conn, level: str) -> None:
    with conn.cursor() as cur:
        cur.execute(f"BEGIN ISOLATION LEVEL {level}")


def one(conn, sql: str):
    with conn.cursor() as cur:
        cur.execute(sql)
        row = cur.fetchone()
        return row[0] if row else None


# ---------------------------------------------------------------- аномалии


def non_repeatable_read(level: str) -> str:
    """T1 читает строку дважды, между чтениями T2 её меняет и фиксирует."""
    setup()
    with psycopg.connect(DSN) as t1, psycopg.connect(DSN, autocommit=True) as t2:
        t1.autocommit = False
        begin(t1, level)
        first = one(t1, "SELECT balance FROM accounts WHERE id = 1")
        fresh(t2, "UPDATE accounts SET balance = 999 WHERE id = 1")
        second = one(t1, "SELECT balance FROM accounts WHERE id = 1")
        t1.rollback()
    return "воспроизвелось" if first != second else "нет"


def phantom(level: str) -> str:
    """T1 дважды считает строки по условию, между счётами T2 добавляет ещё одну."""
    setup()
    with psycopg.connect(DSN) as t1, psycopg.connect(DSN, autocommit=True) as t2:
        t1.autocommit = False
        begin(t1, level)
        first = one(t1, "SELECT count(*) FROM accounts WHERE balance >= 100")
        fresh(t2, "INSERT INTO accounts VALUES (3, 100)")
        second = one(t1, "SELECT count(*) FROM accounts WHERE balance >= 100")
        t1.rollback()
    return "воспроизвелось" if first != second else "нет"


def write_skew(level: str) -> str:
    """Классика: два дежурных врача, каждый снимает с дежурства себя.

    Инвариант — «на дежурстве остаётся хотя бы один». Каждая транзакция его
    проверяет ПЕРЕД записью и видит двоих, поэтому считает свой уход законным.
    Ни одна не трогает строку другой, поэтому конфликта записи нет — и на
    снимке обе проходят. Инвариант при этом нарушен: дежурных ноль.
    """
    setup()
    with psycopg.connect(DSN) as t1, psycopg.connect(DSN) as t2:
        t1.autocommit = t2.autocommit = False
        begin(t1, level)
        begin(t2, level)

        # Обе читают инвариант ДО того, как хоть одна записала.
        seen1 = one(t1, "SELECT count(*) FROM oncall WHERE on_call")
        seen2 = one(t2, "SELECT count(*) FROM oncall WHERE on_call")
        if seen1 < 2 or seen2 < 2:
            return f"негодная подготовка: {seen1}/{seen2}"

        fresh(t1, "UPDATE oncall SET on_call = false WHERE doctor = 'Алиса'")
        fresh(t2, "UPDATE oncall SET on_call = false WHERE doctor = 'Борис'")

        t1.commit()
        try:
            t2.commit()
        except ROLLBACK as exc:
            # Здесь отказ приходит именно на фиксации: до неё конфликта нет,
            # строки у транзакций разные. Ср. lost_update, где он приходит
            # раньше — на самой команде.
            return f"остановлено на фиксации, {exc.sqlstate}"

    with psycopg.connect(DSN, autocommit=True) as check:
        left = one(check, "SELECT count(*) FROM oncall WHERE on_call")
    return "воспроизвелось, дежурных 0" if left == 0 else f"нет, дежурных {left}"


def lost_update(level: str) -> str:
    """Обе транзакции читают баланс и пишут своё значение поверх."""
    setup()
    with psycopg.connect(DSN) as t1, psycopg.connect(DSN) as t2:
        t1.autocommit = t2.autocommit = False
        begin(t1, level)
        begin(t2, level)
        b1 = one(t1, "SELECT balance FROM accounts WHERE id = 1")
        b2 = one(t2, "SELECT balance FROM accounts WHERE id = 1")

        fresh(t1, f"UPDATE accounts SET balance = {b1 + 10} WHERE id = 1")
        t1.commit()

        # Место отказа фиксируется отдельно: он приходит на UPDATE, а не на
        # COMMIT, — это и есть причина, по которой повторять надо транзакцию
        # целиком, а не оборачивать одну фиксацию.
        try:
            fresh(t2, f"UPDATE accounts SET balance = {b2 + 20} WHERE id = 1")
        except ROLLBACK as exc:
            return f"остановлено на команде, {exc.sqlstate}"
        try:
            t2.commit()
        except ROLLBACK as exc:
            return f"остановлено на фиксации, {exc.sqlstate}"

    with psycopg.connect(DSN, autocommit=True) as check:
        final = one(check, "SELECT balance FROM accounts WHERE id = 1")
    # Оба прибавления дошли бы до 130. Всё, что меньше, — потерянное обновление.
    return f"воспроизвелось, баланс {final} вместо 130" if final != 130 else "нет"


def dirty_read() -> tuple[str, str]:
    """Грязное чтение на READ UNCOMMITTED: имя уровня и его поведение.

    ПЕРВАЯ РЕДАКЦИЯ ЭТОГО СКРИПТА УТВЕРЖДАЛА, что PostgreSQL подменяет
    запрошенный READ UNCOMMITTED на READ COMMITTED. Запуск это опроверг:
    `transaction_isolation` возвращает ровно `read uncommitted` — имя уровня
    сохраняется. Подменяется не имя, а ПОВЕДЕНИЕ, и проверять надо его:
    пусть соседняя транзакция изменит строку и НЕ зафиксирует, а мы
    попробуем это увидеть.
    """
    setup()
    with psycopg.connect(DSN) as reader, psycopg.connect(DSN) as writer:
        reader.autocommit = writer.autocommit = False

        begin(writer, "READ COMMITTED")
        fresh(writer, "UPDATE accounts SET balance = 777 WHERE id = 1")
        # НЕ фиксируем: ровно это и должно было бы просочиться.

        begin(reader, "READ UNCOMMITTED")
        name = one(reader, "SELECT current_setting('transaction_isolation')")
        seen = one(reader, "SELECT balance FROM accounts WHERE id = 1")
        reader.rollback()
        writer.rollback()

    behaviour = (
        f"воспроизвелось, увидели {seen}"
        if seen == 777
        else f"нет: видно {seen}, то есть значение до незафиксированной правки"
    )
    return name, behaviour


def main() -> None:
    with psycopg.connect(DSN, autocommit=True) as conn:
        version = one(conn, "SHOW server_version")

    print(f"PostgreSQL {version} | что уровень пропускает на самом деле")
    print()

    name, behaviour = dirty_read()
    print("ГРЯЗНОЕ ЧТЕНИЕ НА READ UNCOMMITTED")
    print(f"  transaction_isolation показывает: {name}")
    print(f"  попытка прочитать незафиксированное: {behaviour}")
    print("  То есть имя уровня сохраняется, а поведение — нет: читается")
    print("  зафиксированное. Грязное чтение недостижимо ни на одном уровне.")
    print()

    checks = [
        ("неповторяющееся чтение", non_repeatable_read),
        ("фантом", phantom),
        ("потерянное обновление", lost_update),
        ("write skew", write_skew),
    ]

    width = max(len(name) for name, _ in checks)
    print(f"  {'аномалия':<{width}}  " + "  ".join(f"{lvl:<38}" for lvl in LEVELS))
    for name, fn in checks:
        cells = []
        for level in LEVELS:
            try:
                cells.append(fn(level))
            except Exception as exc:  # noqa: BLE001 — печатаем любой отказ как есть
                cells.append(f"{type(exc).__name__}")
        print(f"  {name:<{width}}  " + "  ".join(f"{c:<38}" for c in cells))

    print()
    print("ЧТО ИЗ ЭТОГО СЛЕДУЕТ")
    print("  Таблица стандарта описывает, что уровню РАЗРЕШЕНО пропускать.")
    print("  PostgreSQL строже её на REPEATABLE READ: фантомы там не проходят,")
    print("  потому что уровень реализован снимком. И при этом write skew —")
    print("  аномалии, которой в таблице стандарта нет вовсе, — он пропускает.")
    print("  Останавливает её SERIALIZABLE — отказом 40001, и в этом сценарии")
    print("  отказ приходит на фиксации. Но не всегда: в строке потерянного")
    print("  обновления тот же 40001 приходит уже на команде. Поэтому повторять")
    print("  надо транзакцию целиком, а не только её фиксацию.")


if __name__ == "__main__":
    try:
        main()
    except psycopg.OperationalError as exc:
        print(f"нет соединения с базой: {exc}", file=sys.stderr)
        print(f"адрес: {DSN}", file=sys.stderr)
        raise SystemExit(1)