#!/usr/bin/env python3
"""Replay one fictional event after a worker exits. Python standard library only."""
import argparse
import hashlib
import json
import os
import platform
import sqlite3
import subprocess
import sys
import tempfile
from datetime import datetime, timezone
from pathlib import Path

HERE = Path(__file__).resolve().parent


def emit(value):
    print(json.dumps(value, ensure_ascii=False), flush=True)


def worker(database, mode, stop, event):
    con = sqlite3.connect(database, isolation_level=None)
    con.execute('PRAGMA synchronous=FULL')
    event_id, job_key = event['event_id'], event['job_key']
    if mode == 'split':
        if con.execute('SELECT 1 FROM processed_events WHERE event_id=?', (event_id,)).fetchone():
            emit({'stage': 'already_processed'})
            con.close()
            return
        con.execute('BEGIN IMMEDIATE')
        con.execute('INSERT INTO follow_up_jobs(event_id,job_key) VALUES (?,?)', (event_id, job_key))
        con.execute('COMMIT')
        emit({'stage': 'job_committed'})
        if stop == 'after_job':
            os._exit(17)
        con.execute('BEGIN IMMEDIATE')
        con.execute('INSERT INTO processed_events(event_id) VALUES (?)', (event_id,))
        con.execute('COMMIT')
    else:
        con.execute('BEGIN IMMEDIATE')
        inserted = con.execute('INSERT OR IGNORE INTO processed_events(event_id) VALUES (?)', (event_id,)).rowcount
        if inserted:
            con.execute('INSERT INTO follow_up_jobs(event_id,job_key) VALUES (?,?)', (event_id, job_key))
        emit({'stage': 'before_commit', 'event_inserted': bool(inserted)})
        if stop == 'before_commit':
            os._exit(17)
        con.execute('COMMIT')
        emit({'stage': 'transaction_committed'})
        if stop == 'after_commit':
            os._exit(17)
    con.close()
    emit({'stage': 'finished'})


def snapshot(database):
    with sqlite3.connect(database) as con:
        return {
            'jobs': [dict(zip(('id', 'event_id', 'job_key'), row)) for row in con.execute('SELECT id,event_id,job_key FROM follow_up_jobs ORDER BY id')],
            'processed_events': [row[0] for row in con.execute('SELECT event_id FROM processed_events ORDER BY event_id')],
        }


def experiment(output):
    event = json.loads((HERE / 'input.json').read_text(encoding='utf-8'))
    output.mkdir(parents=True, exist_ok=False)
    cases = [
        ('split_normal', 'split', 'none', [1, 1], [1, 1]),
        ('split_after_job', 'split', 'after_job', [1, 0], [2, 1]),
        ('atomic_before_commit', 'atomic', 'before_commit', [0, 0], [1, 1]),
        ('atomic_after_commit', 'atomic', 'after_commit', [1, 1], [1, 1]),
    ]
    observations, processes, summaries = [], [], []
    with tempfile.TemporaryDirectory(prefix='event-redelivery-') as temporary:
        for name, mode, stop, expected_first, expected_final in cases:
            database = Path(temporary) / (name + '.sqlite')
            with sqlite3.connect(database) as con:
                journal_mode = con.execute('PRAGMA journal_mode=DELETE').fetchone()[0]
                con.execute('PRAGMA synchronous=FULL')
                synchronous = con.execute('PRAGMA synchronous').fetchone()[0]
                con.execute('CREATE TABLE follow_up_jobs(id INTEGER PRIMARY KEY, event_id TEXT NOT NULL, job_key TEXT NOT NULL)')
                con.execute('CREATE TABLE processed_events(event_id TEXT PRIMARY KEY NOT NULL)')
            counts = []
            for delivery in (1, 2):
                selected_stop = stop if delivery == 1 else 'none'
                process = subprocess.run([sys.executable, str(Path(__file__).resolve()), '--worker', str(database), mode, selected_stop, json.dumps(event)], capture_output=True, text=True, timeout=15)
                expected_exit = 17 if selected_stop != 'none' else 0
                if process.returncode != expected_exit:
                    raise RuntimeError(f'{name}/{delivery}: exit {process.returncode}: {process.stderr}')
                processes.append({'case': name, 'delivery': delivery, 'mode': mode, 'stop': selected_stop, 'exit_code': process.returncode, 'simulated_ack': process.returncode == 0, 'stdout': process.stdout, 'stderr': process.stderr})
                saved = snapshot(database)
                count = [len(saved['jobs']), len(saved['processed_events'])]
                counts.append(count)
                observations.append({'case': name, 'delivery': delivery, **saved})
            passed = counts == [expected_first, expected_final]
            summaries.append({'case': name, 'mode': mode, 'first_stop': stop, 'journal_mode': journal_mode, 'synchronous': synchronous, 'first_counts': counts[0], 'final_counts': counts[1], 'expected_first': expected_first, 'expected_final': expected_final, 'passed': passed})
    for name, records in [('observations.jsonl', observations), ('raw-processes.jsonl', processes)]:
        (output / name).write_text(''.join(json.dumps(r, ensure_ascii=False) + '\n' for r in records), encoding='utf-8')
    summary = {
        'executed_at': datetime.now(timezone.utc).isoformat(),
        'environment': {'python': platform.python_version(), 'sqlite': sqlite3.sqlite_version, 'platform': platform.platform(), 'machine': platform.machine()},
        'input': event,
        'source_sha256': hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
        'input_sha256': hashlib.sha256((HERE / 'input.json').read_bytes()).hexdigest(),
        'cases': summaries,
        'scope': 'Sequential local subprocesses and one SQLite file per case. No broker, LLM, external API, concurrent workers, OS shutdown or power failure.',
        'passed': all(case['passed'] for case in summaries),
    }
    (output / 'results.json').write_text(json.dumps(summary, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
    emit(summary)
    return 0 if summary['passed'] else 1


if __name__ == '__main__':
    if len(sys.argv) > 1 and sys.argv[1] == '--worker':
        worker(sys.argv[2], sys.argv[3], sys.argv[4], json.loads(sys.argv[5]))
    else:
        parser = argparse.ArgumentParser(description=__doc__)
        parser.add_argument('--out', type=Path, default=Path('results-local'), help='New output directory; existing directories are never overwritten')
        args = parser.parse_args()
        raise SystemExit(experiment(args.out))