"""Local SQLite teaching fixture. No network, PostgreSQL, Symfony or real broker.
Creates temporary databases only. Run with Python 3; standard library only.
"""
import os
import sqlite3
import subprocess
import sys
import tempfile
from pathlib import Path


def connect(path):
    return sqlite3.connect(path, isolation_level=None)


def producer(path):
    db = connect(path)
    db.executescript('''
      CREATE TABLE IF NOT EXISTS orders(id TEXT PRIMARY KEY, total INTEGER NOT NULL);
      CREATE TABLE IF NOT EXISTS outbox(
        id TEXT PRIMARY KEY, order_id TEXT NOT NULL, state TEXT NOT NULL,
        token TEXT, until_at INTEGER, attempts INTEGER NOT NULL DEFAULT 0);
    ''')
    return db


def create_order(db, fail=False):
    db.execute('BEGIN')
    try:
        db.execute('INSERT INTO orders VALUES (?,?)', ('C42', 12000))
        db.execute("INSERT INTO outbox(id,order_id,state) VALUES (?,?, 'pending')", ('E42', 'C42'))
        if fail:
            raise RuntimeError('before commit')
        db.execute('COMMIT')
    except Exception:
        db.execute('ROLLBACK')
        raise


def claim(db, token, now):
    return db.execute('''UPDATE outbox SET state='leased',token=?,until_at=?,attempts=attempts+1
       WHERE id='E42' AND (state='pending' OR (state='leased' AND until_at<=?))''',
       (token, now + 10, now)).rowcount == 1


def acknowledge(db, token):
    return db.execute("UPDATE outbox SET state='sent' WHERE id='E42' AND state='leased' AND token=?", (token,)).rowcount == 1


def receive(db, fail=False):
    db.execute('BEGIN')
    try:
        inserted = db.execute("INSERT OR IGNORE INTO inbox VALUES ('E42')").rowcount
        if inserted:
            db.execute('UPDATE effect SET count=count+1')
        if fail:
            raise RuntimeError('consumer before commit')
        db.execute('COMMIT')
    except Exception:
        db.execute('ROLLBACK')
        raise


def run():
    with tempfile.TemporaryDirectory(prefix='outbox-demo-') as folder:
        p = Path(folder) / 'producer.sqlite'
        db = producer(p)
        try:
            create_order(db, fail=True)
        except RuntimeError:
            pass
        observer = connect(p)
        assert observer.execute('SELECT count(*) FROM orders').fetchone()[0] == 0
        assert observer.execute('SELECT count(*) FROM outbox').fetchone()[0] == 0
        observer.close()
        db.close()
        child = subprocess.run([sys.executable, __file__, '--crash-after-commit', str(p)], check=False)
        assert child.returncode == 81
        db = producer(p)
        assert db.execute('SELECT count(*) FROM orders').fetchone()[0] == 1
        assert db.execute("SELECT state FROM outbox WHERE id='E42'").fetchone()[0] == 'pending'
        assert claim(db, 'worker-A', 100)
        assert not claim(db, 'worker-B', 105)
        consumer = connect(Path(folder) / 'consumer.sqlite')
        consumer.executescript('CREATE TABLE inbox(id TEXT PRIMARY KEY); CREATE TABLE effect(count INTEGER); INSERT INTO effect VALUES(0);')
        try:
            receive(consumer, fail=True)
        except RuntimeError:
            pass
        assert consumer.execute('SELECT count(*) FROM inbox').fetchone()[0] == 0
        assert consumer.execute('SELECT count FROM effect').fetchone()[0] == 0
        receive(consumer)  # Simulated delivery; response is lost before producer acknowledgement.
        assert claim(db, 'worker-B', 111)
        assert not acknowledge(db, 'worker-A')  # Old lease cannot close the current attempt.
        receive(consumer)  # Duplicate delivery, same event identity.
        assert acknowledge(db, 'worker-B')
        assert not claim(db, 'worker-C', 200)
        assert consumer.execute('SELECT count FROM effect').fetchone()[0] == 1
        assert consumer.execute('SELECT count(*) FROM inbox').fetchone()[0] == 1
        assert db.execute('SELECT attempts FROM outbox').fetchone()[0] == 2
        # Negative witness: no inbox means the same effect is applied twice.
        naive_effect = 0
        for _ in range(2):
            naive_effect += 1
        assert naive_effect == 2
        consumer.close()
        db.close()
    print('PASS: rollback producer; real process exit after commit; durable pending intent; lease exclusion and recovery; stale acknowledgement refused; consumer rollback; two deliveries, one local effect; naive witness = two effects.')


if __name__ == '__main__':
    if len(sys.argv) == 3 and sys.argv[1] == '--crash-after-commit':
        db = producer(Path(sys.argv[2]))
        create_order(db)
        os._exit(81)
    run()