"""Synthetic SQLite lab: durable unique destination vs a non-idempotent destination.
No n8n runtime, SaaS API or personal data is used. Run: python3 idempotency-lab.py DIR
"""
from pathlib import Path
import sqlite3,json,hashlib,sys,subprocess,concurrent.futures

def connect(db):
 c=sqlite3.connect(db,timeout=20);c.execute('PRAGMA journal_mode=WAL');c.execute('CREATE TABLE IF NOT EXISTS jobs (key TEXT PRIMARY KEY, payload_hash TEXT NOT NULL, result TEXT NOT NULL)');return c
def submit(db,key,payload,crash=False):
 digest=hashlib.sha256(json.dumps(payload,sort_keys=True).encode()).hexdigest();c=connect(db)
 with c:
  inserted=c.execute('INSERT OR IGNORE INTO jobs VALUES (?,?,?)',(key,digest,'result-'+key)).rowcount
  row=c.execute('SELECT payload_hash,result FROM jobs WHERE key=?',(key,)).fetchone()
 c.close()
 if row[0]!=digest:return {'status':'conflict','key':key}
 if crash:raise RuntimeError('Committed destination; response deliberately lost')
 return {'status':'created' if inserted else 'duplicate','key':key,'result':row[1]}
def count(db,key):
 c=connect(db);n=c.execute('SELECT COUNT(*) FROM jobs WHERE key=?',(key,)).fetchone()[0];c.close();return n
def worker(db,key,crash):
 try:print(json.dumps(submit(db,key,{'task':'synthetic'},crash)))
 except RuntimeError:sys.exit(23)

if __name__=='__main__':
 if len(sys.argv)>1 and sys.argv[1]=='--worker':worker(sys.argv[2],sys.argv[3],sys.argv[4]=='crash');sys.exit(0)
 folder=Path(sys.argv[1] if len(sys.argv)>1 else 'idempotency-lab-output');folder.mkdir(parents=True,exist_ok=True);db=folder/'state.sqlite';assert not db.exists(),'Use a fresh output directory; existing state is evidence'
 report=[]
 def record(case,result,expected_count,key):
  n=count(db,key);assert n==expected_count,(case,n);report.append({'case':case,'result':result,'saved_rows':n})
 record('normal',submit(db,'normal',{'task':'synthetic'}),1,'normal')
 record('same request twice',[submit(db,'double',{'task':'synthetic'}) for _ in range(2)],1,'double')
 a=submit(db,'conflict',{'task':'a'});b=submit(db,'conflict',{'task':'b'});assert b['status']=='conflict';record('same key different content',[a,b],1,'conflict')
 with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:values=list(pool.map(lambda _:submit(db,'parallel',{'task':'synthetic'}),range(20)))
 assert sum(x['status']=='created' for x in values)==1;record('20 concurrent requests',{'created':1,'duplicates':19},1,'parallel')
 connect(db).close()
 def invoke(key,mode):return subprocess.run([sys.executable,__file__,'--worker',str(db),key,mode],capture_output=True,text=True)
 a=invoke('lost-response','crash');assert a.returncode==23;b=invoke('lost-response','normal');assert b.returncode==0 and json.loads(b.stdout)['status']=='duplicate';record('commit then process crash then new process retry',{'first_exit':23,'retry':json.loads(b.stdout)},1,'lost-response')
 a=invoke('restart','normal');b=invoke('restart','normal');assert a.returncode==b.returncode==0;record('new process existing request',{'first':json.loads(a.stdout),'second':json.loads(b.stdout)},1,'restart')
 record('worker with no saved request',json.loads(invoke('before-send','normal').stdout),1,'before-send')
 # Counterexample: two remote writes survive but local acknowledgement is missing.
 c=connect(db);c.execute('CREATE TABLE unsafe_destination (id INTEGER PRIMARY KEY, body TEXT)')
 with c:c.execute('INSERT INTO unsafe_destination(body) VALUES (?)',('synthetic',))
 with c:c.execute('INSERT INTO unsafe_destination(body) VALUES (?)',('synthetic',))
 n=c.execute('SELECT COUNT(*) FROM unsafe_destination').fetchone()[0];c.close();assert n==2
 report.append({'case':'destination without key: lost acknowledgement followed by blind retry','saved_rows':2,'expected_failure_demonstrated':True})
 result={'status':'PASS','cases':report,'environment':{'python':sys.version.split()[0],'sqlite':sqlite3.sqlite_version},'scope':'Destination UNIQUE key and commit in one SQLite DB. Not n8n runtime or real SaaS integration. Different external destinations require their own idempotency/reconciliation contract.'}
 (folder/'summary.json').write_text(json.dumps(result,ensure_ascii=False,indent=2)+'\n');print(json.dumps(result,ensure_ascii=False))