#!/usr/bin/env python3
from __future__ import annotations
import hashlib,json,os,shutil,subprocess,sys,tempfile,time,urllib.parse
from pathlib import Path
REQUIRED=['psql','pg_dump','pg_restore']
CONFIRM_ENV='MAKNOW_DAT011_CONFIRM_DISPOSABLE'
PREFIX='maknow_dat011_'
TIMEOUT=45

def parse_local_url(name):
    u=os.environ.get(name,'')
    if not u: raise ValueError(f'{name} missing')
    p=urllib.parse.urlparse(u)
    host=(p.hostname or '').lower(); db=(p.path or '').lstrip('/')
    if p.scheme not in {'postgres','postgresql'}: raise ValueError(f'{name} must be postgres/postgresql URL')
    if host not in {'localhost','127.0.0.1','::1'}: raise ValueError(f'{name} must target local disposable PostgreSQL')
    if not db.startswith(PREFIX): raise ValueError(f'{name} database must start with {PREFIX}')
    if db in {'postgres','template0','template1'}: raise ValueError(f'{name} cannot target shared/system database')
    return u,host,db

def run(cmd,timeout=TIMEOUT):
    return subprocess.run(cmd,text=True,stdout=subprocess.PIPE,stderr=subprocess.PIPE,timeout=timeout,check=False)

def psql(url,sql):
    r=run(['psql','-X','-v','ON_ERROR_STOP=1','-At',url,'-c',sql])
    if r.returncode: raise RuntimeError('psql failed: '+r.stderr.strip())
    return r.stdout.strip()

def sha256(path):
    h=hashlib.sha256();
    with open(path,'rb') as f:
        for b in iter(lambda:f.read(1024*1024),b''):h.update(b)
    return h.hexdigest()

def source_state_ok(url):
    q="""SELECT ((SELECT credits FROM customer_credit WHERE account_id='A-100')=140 AND (SELECT credits FROM customer_credit WHERE account_id='A-200')=70 AND (SELECT credits FROM customer_credit WHERE account_id='A-300')=50 AND (SELECT count(*) FROM customer_credit)=3 AND NOT EXISTS (SELECT 1 FROM customer_credit WHERE note='ACCIDENTAL_BULK_UPDATE'))::text;"""
    return psql(url,q).lower()=='true'

def restore_empty(url):
    q="SELECT count(*) FROM pg_catalog.pg_class c JOIN pg_catalog.pg_namespace n ON n.oid=c.relnamespace WHERE n.nspname NOT IN ('pg_catalog','information_schema') AND n.nspname NOT LIKE 'pg_toast%' AND c.relkind IN ('r','p','v','m','S','f');"
    return psql(url,q)=='0'

def provenance(root):
    rels=['learner/model-contract.json','learner/recovery/logical_restore_plan.template.json','learner/recovery/pitr_plan.template.json','learner/sql/06_verify_pre_destructive_target.sql']
    return {r:sha256(root/r) for r in rels}

def main():
    missing=[x for x in REQUIRED if not shutil.which(x)]
    if '--preflight' in sys.argv:
        status='PASS_STATIC_PREFLIGHT' if not missing else 'NOT_RUN_MISSING_TOOLS'
        print(json.dumps({'schema':'DAT011_LOGICAL_RESTORE_PREFLIGHT_v2','status':status,'missing_tools':missing,'runtime_claimed':False},indent=2))
        return 0 if not missing else 2
    if missing:
        print(json.dumps({'status':'NOT_RUN_MISSING_TOOLS','missing_tools':missing,'actual_postgresql_runtime_evidence_claimed':False})); return 2
    if os.environ.get(CONFIRM_ENV)!='YES':
        print('REFUSED: set MAKNOW_DAT011_CONFIRM_DISPOSABLE=YES only for disposable local targets',file=sys.stderr); return 3
    try:
        source,_,source_db=parse_local_url('MAKNOW_DAT011_SOURCE_URL')
        restore,_,restore_db=parse_local_url('MAKNOW_DAT011_RESTORE_URL')
    except ValueError as e:
        print('REFUSED:',e,file=sys.stderr); return 4
    if source_db==restore_db:
        print('REFUSED: source and restore databases must be distinct',file=sys.stderr); return 5
    if not source_state_ok(source):
        print('REFUSED: source is not at the bounded PRE_DESTRUCTIVE exercise state',file=sys.stderr); return 6
    if not restore_empty(restore):
        print('REFUSED: restore destination must be empty and disposable',file=sys.stderr); return 7
    root=Path(__file__).resolve().parents[2]
    t0=time.monotonic()
    with tempfile.TemporaryDirectory(prefix='maknow_dat011_') as td:
        archive=Path(td)/'logical.custom'
        d=run(['pg_dump','--format=custom','--no-owner','--no-privileges','--file',str(archive),source])
        if d.returncode: print('pg_dump failed',d.stderr,file=sys.stderr); return 8
        listing=run(['pg_restore','--list',str(archive)])
        if listing.returncode or 'customer_credit' not in listing.stdout:
            print('REFUSED: archive listing did not contain bounded fixture',file=sys.stderr); return 9
        rr=run(['pg_restore','--exit-on-error','--single-transaction','--no-owner','--no-privileges','--dbname',restore,str(archive)])
        if rr.returncode: print('pg_restore failed',rr.stderr,file=sys.stderr); return 10
        if not source_state_ok(restore):
            print('FAIL: restored business state does not match PRE_DESTRUCTIVE target',file=sys.stderr); return 11
        schema_fp=psql(restore,"SELECT string_agg(column_name||':'||data_type,',' ORDER BY ordinal_position) FROM information_schema.columns WHERE table_schema='public' AND table_name IN ('customer_credit','recovery_marker');")
        elapsed_ms=round((time.monotonic()-t0)*1000,3)
        receipt={
          'schema':'DAT011_LOGICAL_RESTORE_RUNTIME_RECEIPT_v2','status':'LOGICAL_RESTORE_VERIFIED',
          'source_database':source_db,'restore_database':restore_db,'source_restore_distinct':True,
          'archive_sha256':sha256(archive),'archive_list_contains_fixture':True,
          'business_state_verified':True,'schema_fingerprint':schema_fp,'measured_restore_time_ms':elapsed_ms,
          'postgresql_version':psql(restore,'SHOW server_version;'),'provenance_sha256':provenance(root),
          'actual_postgresql_runtime_evidence_claimed':True,'actual_pitr_runtime_evidence_claimed':False,
          'fabricated_runtime_evidence':False,'cleanup_status':'TEMP_ARCHIVE_REMOVED_BY_TEMPORARY_DIRECTORY'
        }
        print(json.dumps(receipt,indent=2))
    return 0
if __name__=='__main__': raise SystemExit(main())