import json,subprocess,hashlib,time,collections,os
from pathlib import Path
from mysql_encoding import PROBE_SQL,validate_encoding,consume_probe
from data_contract import canonical_row,quote_identifier,select_expression
r=Path(__file__).resolve().parents[1]
b=json.loads((r/'private/mysql-original-baseline.json').read_text());m=json.loads((r/'private/original-export-manifest.json').read_text());cols=collections.defaultdict(list)
validate_encoding(b.get('encoding'));validate_encoding(m.get('encoding'))
for c in b['columns']:cols[c['TABLE_NAME']].append(c)
queries=[PROBE_SQL,"SET SESSION time_zone='+08:00'; START TRANSACTION WITH CONSISTENT SNAPSHOT;"]
for t in b['tables']:
 name=t['TABLE_NAME'];expr=','.join(select_expression(c)for c in cols[name]);queries.append("SELECT JSON_OBJECT('table','"+name+"','row',JSON_ARRAY("+expr+")) FROM "+quote_identifier(name)+';')
queries.append('COMMIT;')
cmd=['docker','exec','-i','foodsafe2-mysql8-stage-20261023','mysql','--defaults-extra-file=/run/secrets/client.cnf','--default-character-set=utf8mb4','--database=foodsafe2','--batch','--raw','--quick','--skip-column-names','--max-allowed-packet=1G']
log=(r/'private/post-upgrade-verify.stderr').open('wb');p=subprocess.Popen(cmd,stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=log);p.stdin.write('\n'.join(queries).encode('utf8'));p.stdin.close()
encoding=consume_probe(p)
counts={k:0 for k in b['counts']};hashes=collections.defaultdict(list);t0=time.time();last=t0;total=0
for line in p.stdout:
 obj=json.loads(line);name=obj['table'];row=obj['row'];counts[name]+=1;total+=1;hashes[name].append(hashlib.sha256(canonical_row(row)).digest())
 if time.time()-last>=20:print(json.dumps({'phase':'verify_original_preserved','table':name,'rows':total,'elapsed_s':round(time.time()-t0)}),flush=True);last=time.time()
rc=p.wait();log.close()
if rc:raise SystemExit('Verification source query failed')
diff={}
for name in counts:
 actual=hashlib.sha256(b''.join(sorted(hashes[name]))).hexdigest()
 if counts[name]!=int(b['counts'][name])or actual!=m['multiset_sha256'][name]:diff[name]={'count_match':counts[name]==int(b['counts'][name]),'digest_match':actual==m['multiset_sha256'][name]}
res={'encoding':encoding,'original_tables_verified':len(counts),'original_rows_verified':total,'original_columns_verified':len(b['columns']),'row_count_and_full_row_digest_differences':diff,'unchanged':not diff,'elapsed_s':round(time.time()-t0),'includes_full_json_and_base64_values':True}
(r/'evidence/original-preservation.json').write_text(json.dumps(res,indent=2));print(json.dumps(res),flush=True)
if diff:raise SystemExit('Original values changed unexpectedly')