"""Run real SQLite rollout faults and explicit theoretical PostgreSQL lock cases.

Every database lives inside a fresh temporary root. Child processes pause at a
transaction boundary so the parent can issue a real SIGKILL reproducibly.
"""
from __future__ import annotations
import csv, hashlib, html, json, platform, selectors, signal, sqlite3, subprocess
import sys, tempfile, threading, time
from pathlib import Path
import rehearse as recipe
ROOT=Path(__file__).resolve().parent

def digest(path):return hashlib.sha256(Path(path).read_bytes()).hexdigest()

def ready(proc):
    with selectors.DefaultSelector() as selector:
        selector.register(proc.stdout,selectors.EVENT_READ)
        if not selector.select(15):
            proc.kill();proc.wait();raise RuntimeError('Child did not reach transaction checkpoint')
    line=proc.stdout.readline()
    if not line:raise RuntimeError('Child ended before checkpoint: '+proc.stderr.read())
    return json.loads(line)

def child(mode,path):
    db=recipe.connect(path);original=recipe.atomic;calls=0
    def pause(label):
        print(json.dumps({'mode':mode,'boundary':label,'checkpoint':int(recipe.metadata(db,'checkpoint'))}),flush=True)
        sys.stdin.readline()
    def instrument(connection,fn):
        nonlocal calls
        calls+=1
        if mode=='backfill-after-commit':
            value=original(connection,fn)
            if calls==3:pause('after-third-commit')
            return value
        if (mode=='backfill-before-commit' and calls==4) or mode=='contract-before-commit':
            connection.execute('BEGIN IMMEDIATE')
            value=fn()
            pause('transaction-written-before-commit')
            connection.commit();return value
        return original(connection,fn)
    recipe.atomic=instrument
    if mode=='contract-before-commit':recipe.contract(db)
    else:recipe.backfill(db,250)
    db.close()

def run_child(mode,path):
    proc=subprocess.Popen([sys.executable,__file__,'--child',mode,str(path)],stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True)
    try:
        msg=ready(proc);proc.kill();proc.wait(timeout=10)
        if proc.returncode!=-signal.SIGKILL:raise RuntimeError('Child did not die from SIGKILL')
        return msg
    finally:
        if proc.poll() is None:proc.kill();proc.wait()
        for s in (proc.stdin,proc.stdout,proc.stderr):s.close()

def emit(checks,observations,started):
    result=ROOT/'results/operations';result.mkdir(parents=True,exist_ok=True)
    limits=[
        'Crash tests kill the Python process on Linux while SQLite retains normal filesystem and kernel behavior. They do not simulate power loss, torn sectors or storage-controller failures.',
        'PostgreSQL lock and failed-index cases are deterministic theoretical state models informed by PostgreSQL documentation, not measured PostgreSQL sessions.',
        'The concurrent writer is a synthetic workload over one asset table. No child-table migration, service discovery or live customer traffic is claimed.',
        'The original tag is recovered by an executed forward migration from an exact pre-contract tag archive. Without that archive, capitalization and whitespace cannot be reconstructed.',
        'Local sizes and timings are observations of this environment, not production capacity or downtime guarantees.'
    ]
    data={'schema_version':1,'evidence_type':'executed synthetic operations','command':'python operations.py','passed':all(c['passed'] for c in checks),'checks':checks,'source_sha256':{name:digest(ROOT/name) for name in ['operations.py','rehearse.py']},'report':'operations-report.html','limitations':limits,'runtime':{'python':platform.python_version(),'sqlite':sqlite3.sqlite_version,'platform':platform.system()},'observations':observations,'seconds':round(time.perf_counter()-started,6)}
    (ROOT/'completion.json').write_text(json.dumps(data,indent=2)+'\n');(result/'results.json').write_text(json.dumps(data,indent=2)+'\n')
    with (result/'acceptance.csv').open('w',newline='') as f:
        w=csv.DictWriter(f,fieldnames=['id','description','method','expected','observed','passed']);w.writeheader();w.writerows({k:json.dumps(v,sort_keys=True) if isinstance(v,(list,dict)) else v for k,v in c.items()} for c in checks)
    esc=lambda x:html.escape(str(x))
    rows=''.join('<tr>'+''.join('<td>'+esc(c[k])+'</td>' for k in ['id','description','method','expected','observed'])+'<td>'+('Matched' if c['passed'] else 'Mismatch')+'</td></tr>' for c in checks)
    page=f'''<!doctype html><html lang="en"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1"><title>Migration operational rehearsal</title><link rel="stylesheet" href="report.css"></head><body><main><p class="eyebrow">Executed rollout and theoretical database review</p><h1>Migration operational rehearsal</h1><p>{sum(c['passed'] for c in checks)}/{len(checks)} expected outcomes matched. This record separates real SQLite behavior from the modeled PostgreSQL rollout.</p><h2>Acceptance record</h2><div class="table-wrap"><table><thead><tr><th>Case</th><th>Requirement</th><th>Method</th><th>Expected</th><th>Observed</th><th>Check</th></tr></thead><tbody>{rows}</tbody></table></div><h2>Recovery behavior</h2><p>Backfill survives a process kill both between committed batches and inside an uncommitted batch. Killing contraction after its table rebuild but before commit restores the previous schema on reconnect. A separate forward-recovery exercise adds the original tag back from a retained archive while preserving a post-contract description update.</p><h2>Recorded workload and model inputs</h2><pre>{esc(json.dumps(observations,indent=2))}</pre><h2>Evidence boundaries</h2><ul>{''.join('<li>'+esc(x)+'</li>' for x in limits)}</ul><p><a href="completion.json">Verification record</a> · <a href="results/operations/acceptance.csv">Acceptance CSV</a> · <a href="OPERATIONS.md">Runbook and PostgreSQL review</a> · <a href="operations.py">Reproduction source</a></p></main></body></html>'''
    (ROOT/'operations-report.html').write_text(page)
    print(json.dumps({'checks':len(checks),'matched':sum(c['passed'] for c in checks),'passed':data['passed'],'report':'operations-report.html'}))
    return 0 if data['passed'] else 1

def main():
    started=time.perf_counter();checks=[];observations={}
    def record(id,description,method,expected,observed,passed):checks.append(dict(id=id,description=description,expected=expected,observed=observed,passed=bool(passed),method=method))
    with tempfile.TemporaryDirectory(prefix='migration-operations-') as tmp:
        root=Path(tmp)
        for mode in ['backfill-after-commit','backfill-before-commit','contract-before-commit']:
            path=root/(mode+'.sqlite');recipe.create_fixture(path,5000);db=recipe.connect(path);before=recipe.content_digest(db);recipe.expand(db)
            if mode.startswith('contract'):
                recipe.backfill(db,250);recipe.cutover(db)
            db.close();boundary=run_child(mode,path);db=recipe.connect(path)
            integrity=db.execute('PRAGMA integrity_check').fetchone()[0]
            recovered=recipe.profile(db);columns=[r['name'] for r in db.execute('PRAGMA table_info(assets)')]
            if mode.startswith('backfill'):
                persisted=db.execute('SELECT COUNT(*) FROM assets WHERE asset_key IS NOT NULL').fetchone()[0]
                good_state=persisted==750 and recovered['checkpoint']==750 and recovered['stage']=='expanded'
                resumed=recipe.backfill(db,250);recipe.cutover(db);recipe.contract(db)
                expected='750 committed rows retained; uncommitted batch absent; 4250 rows resume; integrity ok'
                observed={'persisted_rows':persisted,'persisted_checkpoint':recovered['checkpoint'],'resume_rows':resumed['updated_rows'],'integrity':integrity,'kill_boundary':boundary}
            else:
                good_state=recovered['stage']=='cutover' and 'tag' in columns and db.execute("SELECT COUNT(*) FROM sqlite_master WHERE type='trigger' AND name LIKE 'assets_legacy_%'").fetchone()[0]==2
                recipe.contract(db);expected='pre-contract schema and triggers restored; clean retry contracts'
                observed={'recovered_stage':recovered['stage'],'legacy_tag_present': 'tag' in columns,'integrity':integrity,'kill_boundary':boundary}
            good_state=good_state and integrity=='ok' and recipe.content_digest(db)==before and recipe.metadata(db,'stage')=='contracted'
            record(mode,'Resume actual recipe after process death','real SIGKILL at transaction boundary',expected,observed,good_state);db.close()
        # Simultaneous old-application writes while the real batched backfill runs.
        path=root/'traffic.sqlite';recipe.create_fixture(path,20000);db=recipe.connect(path);recipe.expand(db);db.close()
        writer_ready=threading.Event();writer_go=threading.Event();writer_done=threading.Event();writer_errors=[];committed=[]
        def writer():
            conn=recipe.connect(path,timeout=5);writer_ready.set();writer_go.wait(10)
            try:
                for i in range(100):
                    key=f' FIELD-{i:06d} '
                    conn.execute('UPDATE assets SET tag=? WHERE id=?',(key,i+1));conn.execute('INSERT INTO assets(id,tag,description) VALUES (?,?,?)',(20001+i,f' NEW-{i:06d} ',f'New field asset {i}'));conn.commit();committed.append(i)
            except Exception as exc:writer_errors.append(repr(exc))
            finally:conn.close();writer_done.set()
        thread=threading.Thread(target=writer);thread.start();writer_ready.wait(10);db=recipe.connect(path,timeout=5)
        # Release the writer after the first actual batch, then allow it to progress.
        atomic=recipe.atomic;batch_calls=0
        def observed_atomic(conn,fn):
            nonlocal batch_calls
            value=atomic(conn,fn);batch_calls+=1
            if batch_calls==1:
                writer_go.set()
                if not writer_done.wait(10):raise RuntimeError('Concurrent writer timed out')
            return value
        recipe.atomic=observed_atomic
        try:fill=recipe.backfill(db,250)
        finally:recipe.atomic=atomic
        thread.join(10)
        consistent=recipe.contracts(db)['consistent'];n=recipe.profile(db)['rows'];nulls=recipe.profile(db)['null_or_empty_keys']
        updates=db.execute("SELECT COUNT(*) FROM assets WHERE id<=100 AND asset_key LIKE 'field-%'").fetchone()[0]
        record('overlap-traffic','Preserve old-application writes during migration','two SQLite connections with a writer thread','100 updates and 100 inserts retained; both contracts agree',{'writer_commits':len(committed),'writer_errors':writer_errors,'rows':n,'normalized_updates':updates,'missing_keys':nulls,'contracts_agree':consistent},len(committed)==100 and not writer_errors and n==20100 and updates==100 and nulls==0 and consistent)
        # Rehearse a reversible recovery route after irreversible schema contraction.
        archive=[list(r) for r in db.execute('SELECT id,tag FROM assets ORDER BY id')]
        archive_path=root/'pre-contract-tags.json'
        archive_path.write_text(json.dumps(archive,separators=(',',':')))
        archive=json.loads(archive_path.read_text())
        archive_digest=hashlib.sha256(json.dumps(archive,separators=(',',':')).encode()).hexdigest()
        recipe.cutover(db);recipe.contract(db);db.execute("UPDATE assets SET description='Post-contract maintenance correction' WHERE id=1");db.commit()
        current_before=recipe.content_digest(db)
        def restore_tag():
            db.execute('ALTER TABLE assets ADD COLUMN tag TEXT')
            db.executemany('UPDATE assets SET tag=? WHERE id=?',((tag,id) for id,tag in archive))
            if db.execute('SELECT COUNT(*) FROM assets WHERE tag IS NULL').fetchone()[0]:raise RuntimeError('Archive omitted current assets')
        recipe.atomic(db,restore_tag)
        actual=[list(r) for r in db.execute('SELECT id,tag FROM assets ORDER BY id')]
        actual_digest=hashlib.sha256(json.dumps(actual,separators=(',',':')).encode()).hexdigest()
        record('forward-recovery','Recover removed source tags without losing newer content','executed forward schema change using pre-contract archive','exact original tags and post-contract description retained',{'archived_rows':len(archive),'archive_sha256':archive_digest,'restored_sha256':actual_digest,'newer_content_preserved':recipe.content_digest(db)==current_before},archive_digest==actual_digest and recipe.content_digest(db)==current_before)
        db.close()
        # Fail a real write-lock acquisition, then retry after release.
        path=root/'lock.sqlite';recipe.create_fixture(path,5000);holder=recipe.connect(path);holder.execute('BEGIN IMMEDIATE');contender=recipe.connect(path,timeout=.075);start=time.perf_counter();error=None
        try:recipe.expand(contender)
        except sqlite3.OperationalError as exc:error=str(exc)
        waited=time.perf_counter()-start;unmodified=recipe.metadata(contender,'stage')=='legacy';holder.rollback();holder.close();recipe.expand(contender);recovered=recipe.metadata(contender,'stage')=='expanded';contender.close()
        record('lock-abort-retry','Timeout leaves the schema unchanged and a later attempt succeeds','real SQLite held write transaction','locked error; legacy schema retained; retry expands',{'error':error,'observed_wait_seconds':round(waited,6),'timeout_seconds':.075,'unchanged_on_timeout':unmodified,'retry_succeeded':recovered},error is not None and 'locked' in error and waited>=.05 and unmodified and recovered)
        # Local size and batch sensitivity measurements. No invented throughput target.
        scale=[]
        for rows,batch in [(5000,100),(5000,1000),(50000,100),(50000,1000),(100000,1000)]:
            path=root/f'scale-{rows}-{batch}.sqlite';recipe.create_fixture(path,rows);db=recipe.connect(path);before=recipe.content_digest(db);t=time.perf_counter();recipe.expand(db);fill=recipe.backfill(db,batch);recipe.cutover(db);recipe.contract(db);elapsed=time.perf_counter()-t
            retained=recipe.profile(db)['rows']==rows and recipe.content_digest(db)==before and db.execute('PRAGMA integrity_check').fetchone()[0]=='ok'
            size=sum(p.stat().st_size for p in root.glob(path.name+'*'));db.close()
            scale.append({'rows':rows,'batch_rows':batch,'seconds':round(elapsed,6),'rows_per_second_observed':round(rows/elapsed,1),'database_and_wal_bytes':size,'content_preserved':retained,'committed_batches':fill['committed_batches']})
        record('size-and-batch-sweep','Preserve contents across fixture size and batch sizes','five executed SQLite rollouts','all row counts, content digests and integrity checks match',{'cases':len(scale),'largest_rows':100000,'all_preserved':all(x['content_preserved'] for x in scale)},all(x['content_preserved'] for x in scale));observations['size_sweep']=scale
        # PostgreSQL model: declared subset of lock modes relevant to this recipe.
        conflicts={
            'AccessShare':{'AccessExclusive'},
            'RowExclusive':{'Share','ShareRowExclusive','Exclusive','AccessExclusive'},
            'ShareUpdateExclusive':{'ShareUpdateExclusive','Share','ShareRowExclusive','Exclusive','AccessExclusive'},
            'Share':{'RowExclusive','ShareUpdateExclusive','ShareRowExclusive','Exclusive','AccessExclusive'},
            'AccessExclusive':{'AccessShare','RowExclusive','ShareUpdateExclusive','Share','AccessExclusive'}
        }
        scenarios=[('ddl-reader','AccessShare','AccessExclusive',2.0,.25,'timeout'),('index-writer','RowExclusive','Share',.8,2.0,'granted_after_wait'),('concurrent-index-writer','RowExclusive','ShareUpdateExclusive',.8,.25,'granted'),('two-index-builds','ShareUpdateExclusive','ShareUpdateExclusive',2.0,.25,'timeout')]
        model=[]
        for name,held,requested,held_until,timeout,expected in scenarios:
            blocked=requested in conflicts[held] or held in conflicts[requested]
            wait=min(held_until,timeout) if blocked else 0
            result='timeout' if blocked and held_until>timeout else 'granted_after_wait' if blocked else 'granted'
            model.append({'scenario':name,'held':held,'requested':requested,'holder_release_s':held_until,'lock_timeout_s':timeout,'modeled_wait_s':wait,'outcome':result,'expected':expected})
        record('postgres-lock-model','Evaluate DDL/index contention under declared PostgreSQL lock modes','theoretical lock compatibility and timeline model','DDL times out; regular index waits; concurrent index admits writer; concurrent builders conflict',{'modeled_cases':len(model),'matched':sum(x['outcome']==x['expected'] for x in model)},all(x['outcome']==x['expected'] for x in model));observations['postgresql_lock_model']=model
        # Concurrent index failure / safe retry cannot infer validity from a name.
        index={'exists':False,'valid':False};trace=[]
        for action in ['start_build','duplicate_detected','inspect_catalog','drop_invalid','repair_fixture','retry_build','validate_catalog']:
            if action=='start_build':index={'exists':True,'valid':False}
            elif action=='duplicate_detected':index['valid']=False
            elif action=='drop_invalid':index={'exists':False,'valid':False}
            elif action=='retry_build':index={'exists':True,'valid':True}
            trace.append({'action':action,**index,'release_allowed':index['exists'] and index['valid'] and action=='validate_catalog'})
        record('postgres-invalid-index-model','Block cutover until a repaired concurrent index is valid','theoretical failed-build/catalog/retry state model','invalid named index blocks cutover; rebuild becomes valid',{'cutover_allowed_after_failure':trace[2]['release_allowed'],'cutover_allowed_after_validated_retry':trace[-1]['release_allowed']},not trace[2]['release_allowed'] and trace[-1]['release_allowed']);observations['postgresql_index_model']=trace
        # Explicit metadata gate matches brief's nontransactional command boundary.
        plan=[{'stage':'expand','transaction':True},{'stage':'backfill_batch','transaction':True},{'stage':'create_index_concurrently','transaction':False},{'stage':'validate_index','transaction':True},{'stage':'contract','transaction':True}]
        valid=next(p for p in plan if p['stage']=='create_index_concurrently')['transaction'] is False
        record('postgres-stage-boundary','Keep concurrent index creation outside a transaction block','theoretical executor schedule','concurrent index stage is standalone',plan,valid);observations['postgresql_stage_plan']=plan
    return emit(checks,observations,started)

if __name__=='__main__':
    if len(sys.argv)>1 and sys.argv[1]=='--child':child(*sys.argv[2:])
    else:raise SystemExit(main())