"""Run ledger process-crash, concurrent ingestion, close and recovery drills."""
import concurrent.futures, hashlib, http.client, json, os, shutil, sqlite3, subprocess, sys, tempfile, threading, time
from pathlib import Path
from app import make_server, DEMO_TOKENS
from ledger import Ledger
from operations import backup, restore, inspect_database
from operations_report import Record
ROOT=Path(__file__).resolve().parent

def event(eid='process-event',quantity='1.125'):
 return {'source':'synthetic-gateway','event_id':eid,'unit':'compute_minute','quantity':quantity,'occurred_at':'2026-01-15T12:00:00Z'}
def child(mode,path):
 l=Ledger(path)
 if mode in ('--crash-ingest','--crash-close'):
  original=l.audit
  def audit(db,account,actor,action,subject,detail=''):
   original(db,account,actor,action,subject,detail)
   if action==('ingest' if mode=='--crash-ingest' else 'period_closed'):os._exit(86)
  l.audit=audit
  if mode=='--crash-ingest':l.ingest('demo',event())
  else:l.close('demo','2026-01')
 else:l.ingest('demo',event());os._exit(87)
def req(s,path,body=None,headers=None):
 c=http.client.HTTPConnection('127.0.0.1',s.server_port,timeout=4);c.request('POST' if body is not None else 'GET',path,json.dumps(body) if body is not None else None,{'Content-Type':'application/json','Authorization':'Bearer demo-operator-token',**(headers or {})});response=c.getresponse();data=response.read();c.close();return response.status,json.loads(data)
def main():
 r=Record('Usage ledger operations rehearsal')
 with tempfile.TemporaryDirectory() as td:
  home=Path(td);path=home/'crash.sqlite';l=Ledger(path)
  run=subprocess.run([sys.executable,__file__,'--crash-ingest',str(path)],capture_output=True,timeout=15)
  facts=inspect_database(path)
  r.check('crash.ingest','Abrupt exit before ingest commit leaves every business table empty',{'exit':86,'rows':0},{'exit':run.returncode,'rows':sum(facts['table_counts'].values())},'Child os._exit after event, entry, posting, totals and audit writes but before SQLite commit.')
  run=subprocess.run([sys.executable,__file__,'--crash-committed',str(path)],capture_output=True,timeout=15);l=Ledger(path);retry=l.ingest('demo',event())
  r.check('crash.response','Committed ingest survives process exit and replay creates no extra quantity',{'exit':87,'duplicate':True,'quantity':'1.125'},{'exit':run.returncode,'duplicate':retry['duplicate'],'quantity':l.reconcile('demo','2026-01')['rebuilt']['compute_minute']['quantity']},'Child exits after commit. Parent reopens the database and retries the original identity.')
  path=home/'load.sqlite';l=Ledger(path);unique=400;attempts=[i for i in range(unique) for _ in range(2)]
  def ingest(i):return Ledger(path).ingest('demo',event('load-'+str(i)))
  started=time.perf_counter()
  with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:results=list(pool.map(ingest,attempts))
  elapsed=time.perf_counter()-started;recon=l.reconcile('demo','2026-01')
  r.check('load.idempotency','Eight concurrent writers turn 800 requests into 400 source events',{'new':400,'duplicate':400,'matches':True},{'new':sum(not x['duplicate'] for x in results),'duplicate':sum(x['duplicate'] for x in results),'matches':recon['matches']},'Separate SQLite connections ingest every synthetic event identity twice.')
  r.check('load.exact-total','Concurrent exact quantities sum without rounding','450.000',recon['rebuilt']['compute_minute']['quantity'],'400 unique events x 1.125 minutes = 450.000 minutes using integer base units.')
  for i in range(10):l.correct('demo',{'source':'synthetic-adjustment','event_id':f'correction-{i}','original_source':'synthetic-gateway','original_event_id':f'load-{i}','replacement_quantity':'0.875','reason':'Synthetic source correction'})
  r.check('load.corrections','Ten corrections append reversals and replacements with the expected net','447.500',l.reconcile('demo','2026-01')['rebuilt']['compute_minute']['quantity'],'450.000 - 10 x (1.125 - 0.875) = 447.500 minutes.')
  r.metrics['concurrent_ingestion']={'writers':8,'requests':800,'unique_events':400,'elapsed_seconds':round(elapsed,4),'requests_per_second':round(800/elapsed,2),'qualification':'Finite local synthetic run with repeated connection initialization, not a production rate guarantee.'}
  run=subprocess.run([sys.executable,__file__,'--crash-close',str(path)],capture_output=True,timeout=15);state=Ledger(path).state('demo')
  r.check('close.rollback','Crash before close commit leaves period open and no frozen version',{'exit':86,'status':'open','closes':0},{'exit':run.returncode,'status':state['periods'][0]['status'],'closes':len(state['closes'])},'Child exits after writing the close, period status and audit row but before commit.')
  l.close('demo','2026-01');frozen=l.export('demo','2026-01');snapshot=backup(path,home/'snapshot')
  late=event('late-after-snapshot','3.125');correction={'source':'synthetic-adjustment','event_id':'late-correction','original_source':'synthetic-gateway','original_event_id':'load-10','replacement_quantity':'0.625','reason':'Synthetic correction after close'}
  l.ingest('demo',late);l.correct('demo',correction)
  r.check('close.immutability','Late event and correction leave the frozen January digest unchanged',frozen['sha256'],l.export('demo','2026-01')['sha256'],'Post-close input stays pending. Export rebuild uses the stored posting cutoff.')
  server=make_server(l,DEMO_TOKENS,0);thread=threading.Thread(target=server.serve_forever,daemon=True);thread.start()
  try:
   r.check('api.isolation','Other-account read and viewer mutation are rejected',[403,403],[req(server,'/api/accounts/other/state')[0],req(server,'/api/accounts/demo/events',event('forbidden'),{'Authorization':'Bearer demo-viewer-token'})[0]],'Real HTTP requests use a demo-scoped operator token and a viewer token.')
   r.check('api.boundary','Foreign Host and Origin are rejected',[403,403],[req(server,'/api/accounts/demo/state',headers={'Host':'evil.invalid'})[0],req(server,'/api/accounts/demo/state',headers={'Origin':'https://evil.invalid'})[0]],'Otherwise authorized real loopback requests.')
   original=l.ingest
   def unavailable(*args,**kwargs):raise sqlite3.OperationalError('Injected full disk')
   l.ingest=unavailable
   try:status,body=req(server,'/api/accounts/demo/events',event('failed-storage'))
   finally:l.ingest=original
   r.check('api.storage','Storage failure returns HTTP 503 without inventing success',503,status,'Inject sqlite3.OperationalError before domain ingestion; retry keeps the stable event identity.')
  finally:server.shutdown();server.server_close();thread.join()
  started=time.perf_counter();restored=restore(home/'snapshot',home/'restored');restore_time=time.perf_counter()-started;recovered=Ledger(restored['database'])
  r.check('restore.frozen-export','Restored January export matches every frozen row and digest',{'sha256':frozen['sha256'],'entries':len(frozen['entries']),'identical':True},{'sha256':recovered.export('demo','2026-01')['sha256'],'entries':len(recovered.export('demo','2026-01')['entries']),'identical':frozen==recovered.export('demo','2026-01')},'Restore the pre-late-input snapshot into a new directory and rebuild the canonical export.')
  r.check('restore.point-in-time','Snapshot correctly excludes the two records received after backup',0,recovered.state('demo')['pending_count'],'This expected loss demonstrates a nonzero backup recovery-point gap.')
  replay=[{'operation':'ingest','data':late},{'operation':'correct','data':correction}]
  (ROOT/'results/operations-replay.json').write_text(json.dumps(replay,indent=2)+'\n')
  for item in replay:getattr(recovered,item['operation'])('demo',item['data'])
  for item in replay:getattr(recovered,item['operation'])('demo',item['data'])
  r.check('restore.source-replay','Stable upstream identities restore two lost records exactly once',2,recovered.state('demo')['pending_count'],'Replay the retained synthetic input journal twice against the restored snapshot.')
  for item in recovered.state('demo')['pending']:recovered.approve('demo',item['seq'],'2026-02')
  recovered.close('demo','2026-02')
  r.check('restore.carry-forward','Recovered late usage and correction close February at 2.625 minutes','2.625',recovered.export('demo','2026-02')['totals']['compute_minute']['quantity'],'3.125 late minutes - (1.125 - 0.625) correction = 2.625, posted after the immutable January close.')
  r.check('restore.closed-digest','Recovery and carry-forward preserve the January digest',frozen['sha256'],recovered.export('demo','2026-01')['sha256'],'Rebuild January again after February approval and close.')
  shutil.copytree(home/'snapshot',home/'damaged')
  with (home/'damaged/store.sqlite').open('r+b') as f:f.seek(100);f.write(b'BROKEN')
  try:restore(home/'damaged',home/'must-not-exist');rejected=False
  except ValueError:rejected=not (home/'must-not-exist').exists()
  r.check('restore.corruption','Modified snapshot is rejected before destination creation',True,rejected,'Change six bytes without changing checksum manifest.')
  (home/'occupied').mkdir();(home/'occupied/keep').write_text('keep')
  try:restore(home/'snapshot',home/'occupied');preserved=False
  except FileExistsError:preserved=(home/'occupied/keep').read_text()=='keep'
  r.check('restore.no-overwrite','Existing restore destination is preserved',True,preserved,'Attempt recovery into an occupied folder.')
  r.metrics['recovery']={'snapshot_bytes':snapshot['bytes'],'snapshot_seconds':snapshot['elapsed_seconds'],'restore_seconds':round(restore_time,6),'snapshot_source_events':snapshot['table_counts']['events'],'events_after_snapshot':2,'upstream_replay_operations':4,'new_events_after_replay':2,'nominal_backup_interval_seconds':900,'theoretical_max_unbacked_events_at_2_per_second':1800,'assumption':'At most two events per second, successful backup every 15 minutes. This is a capacity/RPO scenario, not measured customer traffic.'}
  facts=inspect_database(restored['database']);r.check('integrity.final','Recovered database passes integrity and foreign-key checks',{'integrity':'ok','foreign_key_errors':0},{k:facts[k] for k in ['integrity','foreign_key_errors']},'PRAGMA integrity_check and foreign_key_check after replay, approval and close.')
  schedule=subprocess.run([sys.executable,str(ROOT/'backup_schedule.py'),'--db',str(path),'--out',str(home/'scheduled'),'--interval','0.01','--count','2'],capture_output=True,text=True,timeout=15)
  snapshots=list((home/'scheduled').glob('snapshot-*'))
  r.check('backup.schedule','Finite scheduler creates two checked snapshots and stops',{'exit':0,'snapshots':2,'valid':2},{'exit':schedule.returncode,'snapshots':len(snapshots),'valid':sum(inspect_database(p/'store.sqlite')['integrity']=='ok' for p in snapshots)},'Run the supplied CLI in a subprocess with two snapshots and a 10 ms interval. This exercises scheduling without leaving a background process.')
 return 0 if r.write(['app.py','ledger.py','operations.py','operations_rehearsal.py','operations_report.py','backup_schedule.py'],['The crash drill terminates a process before or after commit. It does not simulate disk firmware, physical power loss or media failure.','All usage and tokens are synthetic. No invoice, tax, currency or payment conclusion follows from these quantities.','Point-in-time recovery deliberately loses post-snapshot events until replay. Recovery requires a separately retained source journal with stable identities; the service cannot reconstruct absent source data.','Snapshot checksum verification detects corruption, not malicious replacement of both database and manifest. Protect or encrypt backups where required.','The local timing and 15-minute backup scenario are evidence and assumptions respectively, not a service-level agreement or enabled schedule.']) else 1
if __name__=='__main__':
 if len(sys.argv)>1 and sys.argv[1].startswith('--crash-'):child(*sys.argv[1:])
 else:raise SystemExit(main())