#!/usr/bin/env python3
"""Capture harness for the DDIA Ch. 10 serializable-not-linearizable exercise.

Shows on a REAL PostgreSQL 16 server that SERIALIZABLE isolation is still not
LINEARIZABLE: a SERIALIZABLE transaction can return a STALE read. T2 begins under
SERIALIZABLE and reads x (fixing its snapshot at x=0). T1, a separate session,
then writes x=1 and commits. T2's SECOND read of x STILL returns 0 -- and T2
commits with NO serialization error (no SQLSTATE 40001). The execution is
perfectly serializable (equivalent to running T2 entirely before T1), yet the
read was stale in real time. Linearizability is the recency guarantee that
serializability does not provide. A fresh transaction after T2 commits reads 1:
T1's write was never lost, only invisible to T2's older snapshot.

This is the CAPTURE / VERIFY path, not the reader path. Readers of the exercise
type two `psql` sessions by hand (T1 / T2); this script drives the same two
sessions programmatically so the printed transcript is real, verbatim output
that the write-up quotes.

It is deterministic and idempotent: it drops and recreates its own database
`ex_ch10_serlin` on the shared, already-running server, seeds it, runs the
interleaving, and closes every connection before dropping.

Run:
    uv run --python 3.12 --with "psycopg[binary]" python3 code/serializable_not_linearizable.py

Substrate (already running; this script does NOT manage containers):
    PostgreSQL 16 container `ddia-ch10-pg` on localhost:5432, user=postgres,
    trust auth, maintenance db `study`.
"""

import psycopg

ADMIN_DSN = "host=localhost port=5432 user=postgres dbname=study"
DBNAME = "ex_ch10_serlin"
EX_DSN = f"host=localhost port=5432 user=postgres dbname={DBNAME}"


def recreate_database() -> None:
    """Drop and recreate our own database via an autocommit admin connection."""
    with psycopg.connect(ADMIN_DSN, autocommit=True) as admin:
        admin.execute(f"DROP DATABASE IF EXISTS {DBNAME} WITH (FORCE)")
        admin.execute(f"CREATE DATABASE {DBNAME}")


def seed() -> None:
    """Create the single-register table `reg` and seed (1, 0)."""
    with psycopg.connect(EX_DSN, autocommit=True) as conn:
        conn.execute("DROP TABLE IF EXISTS reg")
        conn.execute("CREATE TABLE reg (id int PRIMARY KEY, x int)")
        conn.execute("INSERT INTO reg VALUES (1, 0)")


def read_x(cur) -> int:
    cur.execute("SELECT x FROM reg WHERE id = 1")
    return cur.fetchone()[0]


def t1_write_x1_and_commit() -> None:
    """T1: a separate session that writes x=1 and commits, in full.

    A fresh connection, its own transaction, committed before it returns --
    exactly what the reader does by hand in session T1.
    """
    with psycopg.connect(EX_DSN, autocommit=False) as t1:
        with t1.cursor() as cur:
            cur.execute("UPDATE reg SET x = 1 WHERE id = 1")
        t1.commit()


def run() -> None:
    print("Single register reg(id=1, x); seeded x = 0.\n")

    print("=== T2: BEGIN ISOLATION LEVEL SERIALIZABLE ===")
    commit_error = None
    with psycopg.connect(EX_DSN, autocommit=False) as t2:
        t2.execute("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE")
        with t2.cursor() as cur:
            # T2's first read fixes its snapshot at x = 0.
            first = read_x(cur)
            print(f"T2  read 1: SELECT x WHERE id=1  ->  {first}")

            # T1 writes x=1 and commits, AFTER T2's snapshot was taken.
            t1_write_x1_and_commit()
            print("T1  committed: UPDATE reg SET x = 1 WHERE id = 1")

            # T2's second read, still in the same SERIALIZABLE transaction.
            second = read_x(cur)
            print(f"T2  read 2: SELECT x WHERE id=1  ->  {second}   (stale in real time)")

        # T2 committed a read-only transaction; expect NO serialization error.
        try:
            t2.commit()
            print("T2  COMMIT: succeeded, no serialization error (no SQLSTATE 40001)")
        except psycopg.errors.SerializationFailure as e:
            commit_error = e.sqlstate
            print(f"T2  COMMIT: FAILED with SQLSTATE {e.sqlstate}")
    print()

    # A fresh transaction, entirely after T2 committed, sees T1's write.
    print("=== T3: a fresh transaction after T2 committed ===")
    with psycopg.connect(EX_DSN, autocommit=False) as t3:
        t3.execute("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE")
        with t3.cursor() as cur:
            fresh = read_x(cur)
            print(f"T3  read: SELECT x WHERE id=1  ->  {fresh}   (T1's write was never lost)")
        t3.commit()
    print()

    print("=== summary ===")
    print(f"T2 read 1 = {first}, T2 read 2 = {second}"
          f"  -> {'STALE READ under SERIALIZABLE' if second == first == 0 else 'unexpected'}")
    print(f"T2 commit = {'FAILED ' + commit_error if commit_error else 'succeeded (no 40001)'}"
          f"  -> serializable order is T2-before-T1, so the stale read is legal")
    print(f"T3 read   = {fresh}"
          f"  -> the write is visible to a later snapshot; only T2's view was stale")


def main() -> None:
    recreate_database()
    seed()
    run()


if __name__ == "__main__":
    main()