"""Execute isolated recovery fault drills and a declared PostgreSQL recovery model.

All databases and processes are disposable synthetic fixtures. No input accepts a
production database or server. The child mode exists only for deterministic kills.
"""
from __future__ import annotations
import concurrent.futures, csv, errno, hashlib, html, io, json, os, platform, selectors
import shutil, signal, sqlite3, subprocess, sys, tarfile, tempfile, threading, time
from pathlib import Path
from unittest.mock import patch
from make_fixtures import create
from restore_check import run, digest
ROOT=Path(__file__).resolve().parent

def child(stage,bundle,work):
    def stop(name,path):
        if name==stage:
            print(json.dumps({'stage':name,'directory':path}),flush=True)
            sys.stdin.readline()
    result=run(Path(bundle),work_parent=Path(work),stage_hook=stop)
    print(json.dumps(result),flush=True)
    return 0 if result['passed'] else 1

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 its stage within 15 seconds')
    line=proc.stdout.readline()
    if not line:raise RuntimeError('Child exited before its stage: '+proc.stderr.read())
    return json.loads(line)

def write_report(checks,observations,started):
    out=ROOT/'results/operations';out.mkdir(parents=True,exist_ok=True)
    sources={str(p.relative_to(ROOT)):digest(p) for p in [ROOT/'operations.py',ROOT/'restore_check.py',ROOT/'make_fixtures.py']}
    limits=[
        'Process termination is real SIGKILL on Linux; it does not remove power or emulate a faulty storage device.',
        'ENOSPC is an injected write error. No host volume was filled.',
        'SQLite restores and consistency probes are executed. The PostgreSQL pipeline and remote transfer timing are explicit theoretical models, not pg_restore or network measurements.',
        'Synthetic records and local timings describe this run only. Backup authenticity needs a trusted signature or protected manifest; hashes alone do not authenticate a backup.',
        'A killed process cannot run its finally block. The supervisor demonstrates cleanup of only the exact child-owned path after confirming process death.'
    ]
    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':sources,'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')
    (out/'results.json').write_text(json.dumps(data,indent=2)+'\n')
    with (out/'acceptance.csv').open('w',newline='') as f:
        writer=csv.DictWriter(f,fieldnames=['id','description','method','expected','observed','passed']);writer.writeheader()
        writer.writerows({k:json.dumps(v,sort_keys=True) if isinstance(v,(dict,list)) else v for k,v in c.items()} for c in checks)
    def esc(x):return html.escape(str(x))
    rows=''.join('<tr><td>'+esc(c['id'])+'</td><td>'+esc(c['description'])+'</td><td>'+esc(c['method'])+'</td><td>'+esc(c['expected'])+'</td><td>'+esc(c['observed'])+'</td><td>'+('Matched' if c['passed'] else 'Mismatch')+'</td></tr>' for c in checks)
    body=f'''<!doctype html><html lang="en"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1"><title>Backup recovery operations</title><link rel="stylesheet" href="report.css"></head><body><main><p class="eyebrow">Executed operations and theoretical recovery</p><h1>Backup recovery operations</h1><p>{sum(c['passed'] for c in checks)}/{len(checks)} expected outcomes matched. A rejected damaged input is a successful detection, not a recovered backup.</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>Interpretation</h2><p>The process-kill checks prove that partial disposable restores do not alter the source backup and can be followed by a clean retry. Cleanup after SIGKILL belongs to the supervising job, because the killed Python process cannot execute its cleanup code. The live-WAL case uses SQLite's backup API; copying only a live database file is not substituted for it.</p><p>The PostgreSQL model carries a backup through download, checksum verification, isolated restore, application probes, report and owned-resource cleanup. Every injected stage failure prevents acceptance. It is a state-transition model of the original deployment design, not a test of PostgreSQL binaries.</p><h2>Operating model</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">Machine-readable record</a> · <a href="results/operations/acceptance.csv">Acceptance CSV</a> · <a href="OPERATIONS.md">Runbook and PostgreSQL analysis</a> · <a href="operations.py">Reproduction source</a></p></main></body></html>'''
    (ROOT/'operations-report.html').write_text(body)
    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='recovery-operations-') as tmp:
        root=Path(tmp);bundle=create(root/'backup');work=root/'owned';work.mkdir();sentinel=work/'unrelated.txt';sentinel.write_text('keep')
        original={p.name:digest(p) for p in bundle.iterdir()}
        # Deterministic process death after a real stage, no Python cleanup opportunity.
        for stage in ['files','restore','probes']:
            proc=subprocess.Popen([sys.executable,__file__,'--child',stage,str(bundle),str(work)],stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True)
            try:
                msg=ready(proc);owned=Path(msg['directory']);proc.kill();proc.wait(timeout=10)
                exited=proc.returncode==-signal.SIGKILL
                leftovers=owned.exists();source_same=all(digest(bundle/n)==sha for n,sha in original.items())
                # The path was emitted by this child; require direct containment and prefix.
                if owned.parent!=work or not owned.name.startswith('recovery-drill-') or owned.is_symlink():raise RuntimeError('Unsafe owned path')
                shutil.rmtree(owned)
                retry=run(bundle,work_parent=work)
                ok=exited and leftovers and source_same and retry['passed'] and sentinel.read_text()=='keep' and sorted(p.name for p in work.iterdir())==['unrelated.txt']
                record('kill-'+stage,'Recover after hard process termination at '+stage,'real subprocess / SIGKILL','partial directory found; source unchanged; supervised cleanup; retry passes',{'exit_signal':-proc.returncode,'partial_directory_observed':leftovers,'source_unchanged':source_same,'retry_passed':retry['passed'],'unrelated_preserved':sentinel.exists()},ok)
            finally:
                if proc.poll() is None:proc.kill();proc.wait()
                for stream in [proc.stdin,proc.stdout,proc.stderr]:stream.close()
        # Explicit simulated storage-full error at the actual archive write function.
        with patch('restore_check.shutil.copyfileobj',side_effect=OSError(errno.ENOSPC,'synthetic full volume')):
            r=run(bundle,work_parent=work)
        record('full-volume','Fail closed and clean owned resources on archive-write failure','injected ENOSPC','rejected; cleanup succeeds',{'passed_restore':r['passed'],'cleanup':r['cleanup_succeeded'],'error':r.get('error')},not r['passed'] and r['cleanup_succeeded'] and 'synthetic full volume' in r.get('error',''))
        # Two workers use their own real database and extraction target.
        with concurrent.futures.ProcessPoolExecutor(max_workers=2) as pool:
            results=list(pool.map(_parallel_run,[(str(bundle),str(work))]*2))
        record('parallel-isolation','Parallel verification does not share a restore target','two operating-system workers','two successful restores; distinct owned directories; source unchanged',{'successes':sum(r['passed'] for r in results),'distinct_directories':len({r['temporary_directory'] for r in results}),'source_unchanged':all(digest(bundle/n)==sha for n,sha in original.items())},all(r['passed'] and r['cleanup_succeeded'] for r in results) and len({r['temporary_directory'] for r in results})==2 and all(digest(bundle/n)==sha for n,sha in original.items()))
        # Last allowed count and first disallowed count, both actual runner paths.
        for count in [1000,1001]:
            fixture=create(root/f'count-{count}',count=count);r=run(fixture,work_parent=work)
            expect=count==1000
            record('count-'+str(count),'Enforce documented file-count boundary','generated SQLite and archive fixture',f'restore_passed={expect}',{'restore_passed':r['passed'],'seconds':r['timings_seconds']['total'],'input_bytes':sum(p.stat().st_size for p in fixture.iterdir()),'error':r.get('error')},r['passed']==expect and r['cleanup_succeeded'])
        # A revised archive-level hash must not hide changed member data.
        damaged=root/'changed-member';shutil.copytree(bundle,damaged)
        with tarfile.open(bundle/'files.tar') as src,tarfile.open(damaged/'files.tar','w') as dst:
            for i,member in enumerate(src):
                data=src.extractfile(member).read()
                if i==0:data=bytes([data[0]^1])+data[1:]
                dst.addfile(member,io.BytesIO(data))
        m=json.loads((damaged/'manifest.json').read_text());m['archive_sha256']=digest(damaged/'files.tar');(damaged/'manifest.json').write_text(json.dumps(m))
        r=run(damaged,work_parent=work)
        record('member-corruption','Reject changed attachment despite updated outer archive checksum','single-byte damage','member digest rejection',r.get('error'),not r['passed'] and 'File digest mismatch' in r.get('error','') and r['cleanup_succeeded'])
        # A consistent online backup while a separate thread commits status changes.
        live=create(root/'live');live_db=live/'database.sqlite';db=sqlite3.connect(live_db);db.execute('PRAGMA journal_mode=WAL');db.close()
        started_writer=threading.Event();allow_writes=threading.Event();during_copy=threading.Event();writer_errors=[]
        def writer():
            try:
                conn=sqlite3.connect(live_db,timeout=10)
                for i in range(300):
                    conn.execute("UPDATE visits SET status=? WHERE id=?",('complete' if i%2 else 'open',i%24+1));conn.commit()
                    if i==0:
                        started_writer.set()
                        if not allow_writes.wait(10):raise RuntimeError('Backup did not release concurrent writer')
                    if i==1:during_copy.set()
                conn.close()
            except Exception as exc:writer_errors.append(repr(exc));started_writer.set()
        thread=threading.Thread(target=writer);thread.start();started_writer.wait(10)
        captured=root/'captured';captured.mkdir();shutil.copyfile(live/'files.tar',captured/'files.tar')
        src=sqlite3.connect(live_db,timeout=10);dst=sqlite3.connect(captured/'database.sqlite')
        overlap_observed=[]
        def progress(status,remaining,total):
            if not allow_writes.is_set():
                allow_writes.set()
                overlap_observed.append(during_copy.wait(10))
        src.backup(dst,pages=1,progress=progress);src.close();dst.close();thread.join(15)
        if thread.is_alive():raise RuntimeError('Synthetic writer did not terminate')
        m=json.loads((live/'manifest.json').read_text());m['database_sha256']=digest(captured/'database.sqlite');(captured/'manifest.json').write_text(json.dumps(m))
        r=run(captured,work_parent=work)
        record('online-wal-snapshot','Recover a snapshot produced while writes commit','SQLite backup API plus independent writer','valid snapshot; 300 commits complete; application probes pass',{'writer_errors':writer_errors,'writer_commits':300 if not writer_errors else None,'restore_passed':r['passed'],'writer_committed_during_copy':overlap_observed==[True]},not writer_errors and r['passed'] and overlap_observed==[True])
        observations['live_snapshot_assumption']='Attachment bytes and document references remain immutable during capture; only visit status changes. A real application must supply an equivalent coordinated snapshot contract.'
        # Declared deterministic recovery state machine for unavailable remote/PostgreSQL stack.
        pipeline=['download','checksum','create_disposable_database','restore','application_probes','report']
        model=[]
        for fault in [None]+pipeline:
            completed=[];created=False;accepted=False
            for step in pipeline:
                if step==fault:break
                completed.append(step)
                if step=='create_disposable_database':created=True
            accepted=len(completed)==len(pipeline)
            cleanup_owned=created
            model.append({'fault':fault or 'none','completed':completed,'accepted':accepted,'cleanup_owned_database':cleanup_owned,'production_target_used':False})
        record('postgres-recovery-model','Never accept a backup after an incomplete required stage','theoretical pipeline state model','healthy path accepted; all six injected faults rejected; no production target',{'scenarios':len(model),'accepted':sum(x['accepted'] for x in model),'failures_rejected':sum(not x['accepted'] for x in model)},model[0]['accepted'] and all(not x['accepted'] for x in model[1:]))
        observations['postgresql_pipeline_model']=model
        # Explicit sizing assumptions; no invented measurement or arbitrary pass threshold.
        scenarios=[]
        for gb in [1,10,100]:
            for mbps in [25,100]:
                transfer=gb*1024/mbps;restore=gb*1024/50;probe=60;startup=30
                scenarios.append({'database_GiB':gb,'network_MiB_s':mbps,'restore_MiB_s_assumed':50,'download_s':transfer,'restore_s':restore,'fixed_startup_and_probe_s':startup+probe,'modeled_RTO_s':transfer+restore+startup+probe,'backup_interval_s':3600,'worst_case_RPO_s':3600})
        record('recovery-budget','Calculate recovery window from declared rates','theoretical serial-stage budget','RTO increases with size and decreases with transfer rate',{'cases':len(scenarios),'range_seconds':[min(x['modeled_RTO_s'] for x in scenarios),max(x['modeled_RTO_s'] for x in scenarios)]},all(scenarios[i]['modeled_RTO_s']>scenarios[i+1]['modeled_RTO_s'] for i in [0,2,4]) and scenarios[0]['modeled_RTO_s']<scenarios[2]['modeled_RTO_s']<scenarios[4]['modeled_RTO_s'])
        observations['capacity_model']=scenarios
        observations['capacity_equation']='RTO = database MiB / download MiB/s + database MiB / restore MiB/s + 30 s startup + 60 s probes. RPO bound assumes successful hourly backups; failed or stale backups invalidate that bound. Compression, parallelism, secrets, retries and operator time are excluded.'
    return write_report(checks,observations,started)

def _parallel_run(args):return run(Path(args[0]),work_parent=Path(args[1]))
if __name__=='__main__':
    if len(sys.argv)>1 and sys.argv[1]=='--child':raise SystemExit(child(*sys.argv[2:]))
    raise SystemExit(main())