"""Python 3 + SQLite 3.35+. In-memory compatibility experiment, not a production load test."""
import json
import platform
import sqlite3

if sqlite3.sqlite_version_info < (3, 35, 0):
    raise SystemExit('SQLite >= 3.35 is required for DROP COLUMN.')

def database():
    db = sqlite3.connect(':memory:')
    db.execute('CREATE TABLE users (id INTEGER PRIMARY KEY, full_name TEXT NOT NULL)')
    db.executemany('INSERT INTO users VALUES (?, ?)', [(1, '민수'), (2, '서연'), (3, '지우')])
    db.commit()
    return db

def old_read(db, user_id):
    return db.execute('SELECT full_name FROM users WHERE id = ?', (user_id,)).fetchone()[0]

def expect_old_read_failure(db):
    try:
        old_read(db, 1)
    except sqlite3.OperationalError as error:
        assert 'no such column' in str(error)
        return True
    raise AssertionError('The old reader should fail after removing full_name.')

naive = database()
naive.execute('ALTER TABLE users RENAME COLUMN full_name TO display_name')
rename_breaks_old_reader = expect_old_read_failure(naive)
naive.close()

db = database()
db.execute('ALTER TABLE users ADD COLUMN display_name TEXT')
assert old_read(db, 1) == '민수'
# Before the writer cutover, old-only writes can still occur. Keep full_name authoritative.
with db:
    db.execute('UPDATE users SET full_name = ? WHERE id = 2', ('서연 수정',))

def compatible_write(user_id, name, fail=False):
    with db:
        db.execute('UPDATE users SET full_name = ?, display_name = ? WHERE id = ?', (name, name, user_id))
        if fail:
            raise RuntimeError('Injected failure before commit')

compatible_write(1, '민수 수정')
try:
    compatible_write(1, '반영되면 안 되는 값', fail=True)
except RuntimeError:
    pass
assert db.execute('SELECT full_name, display_name FROM users WHERE id = 1').fetchone() == ('민수 수정', '민수 수정')

# Gate: all active writers are now compatible. In production verify this outside the DB.
def backfill(batch_size=1):
    total = 0
    while True:
        ids = [row[0] for row in db.execute('SELECT id FROM users WHERE display_name IS NULL ORDER BY id LIMIT ?', (batch_size,))]
        if not ids:
            return total
        with db:
            for user_id in ids:
                cursor = db.execute('UPDATE users SET display_name = full_name WHERE id = ? AND display_name IS NULL', (user_id,))
                total += cursor.rowcount

filled = backfill()
rerun = backfill()
assert filled == 2 and rerun == 0
assert db.execute('SELECT display_name FROM users WHERE id = 2').fetchone()[0] == '서연 수정'
mismatches = db.execute('SELECT COUNT(*) FROM users WHERE display_name IS NULL OR display_name != full_name').fetchone()[0]
assert mismatches == 0
for user_id in (1, 2, 3):
    assert old_read(db, user_id) == db.execute('SELECT display_name FROM users WHERE id = ?', (user_id,)).fetchone()[0]

# Contraction belongs to a separate release, after old readers/writers are retired.
db.execute('ALTER TABLE users DROP COLUMN full_name')
assert db.execute('SELECT display_name FROM users WHERE id = 1').fetchone()[0] == '민수 수정'
old_reader_fails_after_contract = expect_old_read_failure(db)
db.close()
print(json.dumps({
    'experiment': 'expand-migrate-contract reader/writer compatibility',
    'python': platform.python_version(), 'sqlite': sqlite3.sqlite_version,
    'rows': 3, 'renameBreaksOldReader': rename_breaks_old_reader,
    'oldReaderWorksBeforeContract': True, 'dualWriteRollback': 'PASS',
    'backfilledRows': filled, 'rerunChangedRows': rerun, 'mismatches': mismatches,
    'oldReaderFailsAfterContract': old_reader_fails_after_contract,
    'newReaderWorksAfterContract': True, 'assertions': 'PASS',
    'limits': 'Single connection, in-memory SQLite. Does not test production locks, concurrent writers, replication lag or PostgreSQL DDL duration.',
}, ensure_ascii=False, indent=2))