"""Run queue process-crash, signed HTTP recovery, load and restore drills."""
import concurrent.futures, http.client, json, os, shutil, sqlite3, subprocess, sys, tempfile, threading, time
from pathlib import Path
from app import make_server as api_server
from delivery_queue import Queue, DestinationPolicy
from receiver import Receiver, SECRET, make_server
from operations import backup, restore, inspect_database
from operations_report import Record
ROOT=Path(__file__).resolve().parent

def child(path,marker):
 q=Queue(path,clock=lambda:1000.0);job=q.claim(lease_seconds=1);Path(marker).write_text(json.dumps(job));os._exit(86)
def req(server,path,body=None,headers=None):
 c=http.client.HTTPConnection('127.0.0.1',server.server_port,timeout=4);h={'Content-Type':'application/json','Authorization':'Bearer synthetic-administrator-token',**(headers or {})};c.request('POST' if body is not None else 'GET',path,json.dumps(body) if body is not None else None,h);res=c.getresponse();data=res.read();c.close();return res.status,json.loads(data)
def serve(server):
 t=threading.Thread(target=server.serve_forever,daemon=True);t.start();return t
def stop(s,t):s.shutdown();s.server_close();t.join()
def main():
 r=Record('Webhook queue operations rehearsal')
 with tempfile.TemporaryDirectory() as td:
  home=Path(td);path=home/'crash.sqlite';q=Queue(path,DestinationPolicy(approved_hosts=['receiver.example']),clock=lambda:1000.0);q.add_endpoint('receiver','https://receiver.example/webhook',SECRET);q.accept('crash-event','inspection',{'synthetic':True},['receiver'])
  marker=home/'lease.json';run=subprocess.run([sys.executable,__file__,'--crash',str(path),str(marker)],capture_output=True,timeout=15);old=json.loads(marker.read_text());q=Queue(path,clock=lambda:1002.0);new=q.claim(lease_seconds=1)
  r.check('crash.reclaim','Claim survives abrupt exit and a new worker replaces the expired owner',{'exit':86,'same_job':True,'new_token':True,'attempt':2},{'exit':run.returncode,'same_job':new['id']==old['id'],'new_token':new['lease_token']!=old['lease_token'],'attempt':new['attempts']},'Child persists a lease then os._exit. A declared clock advances from 1000 to 1002 seconds, past its one-second lease.')
  stale=q.complete(old['id'],old['lease_token'],http_status=200);current=q.complete(new['id'],new['lease_token'],http_status=200)
  r.check('crash.fence','Old worker completion rejected; replacement completion accepted',[False,True],[stale,current],'Call completion with persisted old and new ownership tokens after restart.')
  receiver=Receiver(home/'receiver.sqlite');server=make_server(receiver,0);thread=serve(server);origin=f'http://127.0.0.1:{server.server_port}';policy=DestinationPolicy(test_origins=[origin]);livepath=home/'live.sqlite';live=Queue(livepath,policy)
  try:
   for name,route in [('normal','ok'),('lost','drop-once')]:live.add_endpoint(name,origin+'/'+route,SECRET,max_inflight=2,max_attempts=4,timeout=1,retry_base=.01)
   for i in range(40):
    for target in ['normal','lost']:live.accept(f'{target}-{i}','inspection',{'sequence':i,'synthetic':True},[target])
   started=time.perf_counter();deadline=time.monotonic()+30
   def work():
    local=Queue(livepath,policy)
    while time.monotonic()<deadline:
     if not local.work_once():
      counts=local.snapshot()['counts']
      if not any(counts.get(k,0) for k in ['ready','retry','leased']):return
      time.sleep(.005)
    raise TimeoutError('Synthetic workload did not drain in 30 seconds')
   with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool:list(pool.map(lambda _:work(),range(4)))
   elapsed=time.perf_counter()-started;state=live.snapshot();receipts=receiver.snapshot()
   r.check('http.delivery','Four workers deliver all 80 accepted jobs',{'delivered':80},state['counts'],'Forty normal and forty first-response-loss events, two in-flight requests per endpoint, real signed loopback HTTP.')
   r.check('http.deduplication','120 requests cause exactly 80 receiver business effects',{'requests':120,'effects':80},{'requests':sum(x['hits'] for x in receipts),'effects':sum(x['effects'] for x in receipts)},'Receiver stores a fingerprint and one effect per endpoint/event identity. Forty first responses are deliberately closed after commit.')
   r.check('http.attempts','Attempt history contains the expected first failures and successful retries',{'retry':40,'delivered':80},{kind:sum(a['outcome']==kind for a in state['attempts']) for kind in ['retry','delivered']},'Count persisted terminal outcomes after every worker has drained.')
   r.metrics['http_workload']={'jobs':80,'workers':4,'http_attempts':120,'elapsed_seconds':round(elapsed,4),'delivered_jobs_per_second':round(80/elapsed,2),'qualification':'A finite local synthetic workload, not internet latency or sustained capacity.'}
   api=api_server(live,'synthetic-administrator-token',0);api_thread=serve(api)
   try:
    r.check('api.authorization','Missing administrator token is rejected',401,req(api,'/api/state',headers={'Authorization':''})[0],'Real loopback HTTP without bearer authorization.')
    r.check('api.boundary','Foreign Host and Origin requests are rejected',[403,403],[req(api,'/api/state',headers={'Host':'evil.invalid'})[0],req(api,'/api/state',headers={'Origin':'https://evil.invalid'})[0]],'Authenticated requests with foreign Host or Origin.')
    original=live.accept
    def unavailable(*args,**kwargs):raise sqlite3.OperationalError('Injected full disk')
    live.accept=unavailable
    try:status,body=req(api,'/api/events',{'id':'storage-failure','type':'inspection','data':{},'endpoints':['normal']})
    finally:live.accept=original
    r.check('api.storage','Storage failure returns retryable HTTP 503 without accepting an event',503,status,'Inject sqlite3.OperationalError at acceptance, before the transaction starts; error response contains no payload or credential.')
   finally:stop(api,api_thread)
   disaster=Queue(home/'recovery-source.sqlite',policy);disaster.add_endpoint('normal',origin+'/ok',SECRET);disaster.accept('restore-replay','inspection',{'synthetic':True},['normal']);snapshot=backup(disaster.path,home/'snapshot')
   disaster.work_once();started=time.perf_counter();restored=restore(home/'snapshot',home/'restored');restore_time=time.perf_counter()-started;recovered=Queue(restored['database'],policy);recovered.work_once()
   receipt=next(x for x in receiver.snapshot() if x['event_id']=='restore-replay')
   r.check('restore.at-least-once','Restoring an older queue snapshot redelivers without duplicate receiver effect',{'requests':2,'effects':1,'status':'delivered'},{'requests':receipt['hits'],'effects':receipt['effects'],'status':recovered.snapshot()['jobs'][0]['status']},'Snapshot a ready job, deliver it, restore the older snapshot into a new directory and deliver again to the same durable receiver.')
   r.check('restore.identity','Restored acceptance still recognizes the same event identity',True,recovered.accept('restore-replay','inspection',{'synthetic':True},['normal'])['duplicate'],'Replay the accepted event through the restored domain API.')
   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','Corrupt snapshot rejected before destination creation',True,rejected,'Change stored 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 directory remains untouched',True,preserved,'Attempt restore into a nonempty directory.')
   r.metrics['recovery']={'snapshot_bytes':snapshot['bytes'],'snapshot_seconds':snapshot['elapsed_seconds'],'restore_seconds':round(restore_time,6),'queue_backup_rpo_seconds':300,'receiver_requirement':'Receiver deduplication records must outlive the longest queue replay or backup retention horizon.'}
   facts=inspect_database(livepath);r.check('integrity.final','Queue integrity and foreign-key checks pass',{'integrity':'ok','foreign_key_errors':0},{k:facts[k] for k in ['integrity','foreign_key_errors']},'Check live workload database after all worker threads stop.')
   # A transparent deterministic deployment model, separate from the measured run.
   attempts=4;timeout=2;backoffs=[min(300,1*2**i) for i in range(attempts-1)];bound=attempts*timeout+sum(backoffs)
   r.metrics['theoretical_outage_envelope']={'attempts':attempts,'per_attempt_seconds':timeout,'retry_delays_seconds':backoffs,'elapsed_if_every_request_uses_timeout_seconds':bound,'excluded_from_bound':['DNS resolver latency','worker scheduling delay','SQLite lock wait'],'arrival_jobs_per_second':2,'outage_seconds':600,'generated_jobs':1200,'conclusion':'A ten-minute outage exceeds this retry envelope. Jobs become dead letters and need deliberate replay; a queue does not guarantee automatic delivery after arbitrary outages.'}
   r.check('theory.retry-budget','Declared four-attempt timeout/backoff calculation totals 15 seconds',15,bound,'Compute 4 x 2 + (1 + 2 + 4); exclude DNS, scheduling and lock waiting explicitly.')
   schedule=subprocess.run([sys.executable,str(ROOT/'backup_schedule.py'),'--db',str(livepath),'--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.')
  finally:stop(server,thread)
 return 0 if r.write(['app.py','delivery_queue.py','receiver.py','operations.py','operations_rehearsal.py','operations_report.py','backup_schedule.py'],['The child crash is real process termination, but lease time is deliberately advanced by an injected clock. It is not machine power-loss testing.','All destinations and events are local synthetic fixtures. No live external destination, DNS outage or public TLS ingress was exercised.','The restore drill preserves the receiver database. Losing receiver deduplication while replaying a queue can duplicate external effects.','Snapshots contain endpoint signing secrets. They require restricted storage and encryption at rest where the environment needs it; the checksum is not a signature.','The theoretical retry calculation excludes unbounded OS DNS timing and scheduler delays. Single-host SQLite provides no automatic failover.']) else 1
if __name__=='__main__':
 if len(sys.argv)>1 and sys.argv[1]=='--crash':child(*sys.argv[2:])
 else:raise SystemExit(main())