"""Isolated SQLite crash/recovery examples; stdlib only. No network requests."""
import sys, json, sqlite3, tempfile, subprocess, os, hashlib
from pathlib import Path
BASE=Path(__file__).resolve().parent

def dbconnect(file):
    db=sqlite3.connect(file)
    db.execute('PRAGMA journal_mode=WAL')
    db.execute('PRAGMA synchronous=FULL')
    return db

def worker(file,kind,payload):
    db=dbconnect(file)
    if kind=='crm':
        d=json.loads(payload)
        db.execute('INSERT INTO records(external_key,payload) VALUES(?,?)',(d['event_key'],json.dumps(d['payload'],ensure_ascii=False,sort_keys=True)))
        db.commit()
    else:
        db.execute("INSERT OR IGNORE INTO results(job_id,body) VALUES(?,?)",(payload,'Результат '+payload));db.commit()
    os._exit(73) # Result committed, acknowledgement deliberately never written.

def automation(directory):
    ds=json.loads((BASE/'deliveries.json').read_text()); dbfile=directory/'crm.sqlite';db=dbconnect(dbfile)
    db.executescript('CREATE TABLE records(external_key TEXT PRIMARY KEY,payload TEXT); CREATE TABLE receipts(event_key TEXT PRIMARY KEY,payload TEXT);')
    trace=[];crash_exit=None
    for d in ds:
        key=d['event_key'];payload=json.dumps(d['payload'],ensure_ascii=False,sort_keys=True)
        row=db.execute('SELECT payload FROM receipts WHERE event_key=?',(key,)).fetchone()
        if row:
            outcome='duplicate' if row[0]==payload else 'payload_conflict';trace.append({'delivery':d['delivery_id'],'key':key,'outcome':outcome});continue
        if not d['payload']['topic'].strip() or d['payload']['category'] not in ['analytics','automation','rag']:
            trace.append({'delivery':d['delivery_id'],'key':key,'outcome':'validation_error'});continue
        stored=db.execute('SELECT payload FROM records WHERE external_key=?',(key,)).fetchone()
        if stored:
            assert stored[0]==payload
            outcome='recovered_by_lookup'
        elif d['test_scenario']=='timeout_after_commit':
            crash_exit=subprocess.run([sys.executable,str(Path(__file__).resolve()),'--worker',str(dbfile),'crm',json.dumps(d)]).returncode
            assert crash_exit==73
            trace.append({'delivery':d['delivery_id'],'key':key,'outcome':'commit_then_process_exit_73'});continue
        else:
            db.execute('INSERT INTO records VALUES(?,?)',(key,payload));outcome='created'
        db.execute('INSERT INTO receipts VALUES(?,?)',(key,payload));db.commit()
        trace.append({'delivery':d['delivery_id'],'key':key,'outcome':outcome})
    records=[{'external_key':x[0],'payload':json.loads(x[1])} for x in db.execute('SELECT * FROM records ORDER BY external_key')]
    assert len(records)==3 and len(trace)==8
    assert any(x['outcome']=='recovered_by_lookup' for x in trace)
    assert next(x for x in records if x['external_key']=='event-002')['payload']['topic']=='Учебная интеграция формы'
    db.close();return {'mode':'local SQLite adapter; no external CRM','crashExit':crash_exit,'deliveries':len(ds),'records':records,'trace':trace,'checks':'PASS'}

def queue(directory):
    source=json.loads((BASE/'backup-fixture.json').read_text());files=json.loads((BASE/'file-payloads.json').read_text())
    dbfile=directory/'queue.sqlite';db=dbconnect(dbfile)
    db.executescript('CREATE TABLE jobs(id TEXT PRIMARY KEY,state TEXT,attempts INTEGER); CREATE TABLE results(job_id TEXT PRIMARY KEY,body TEXT);')
    for j in source['jobs']:
        db.execute('INSERT INTO jobs VALUES(?,?,?)',(j['uuid'],j['status'],j['attempt_count']))
        if j['status']=='completed':db.execute('INSERT INTO results VALUES(?,?)',(j['uuid'],'Результат '+j['uuid']))
    db.commit();before=[dict(zip(['id','state','attempts'],x)) for x in db.execute('SELECT * FROM jobs ORDER BY id')]
    media=[]
    for f in files:
        body=f['content_utf8'].encode(); assert len(body)==f['size_bytes'] and hashlib.sha256(body).hexdigest()==f['sha256']
        path=directory/f['relative_path'];path.parent.mkdir(exist_ok=True);path.write_bytes(body);media.append({'path':f['relative_path'],'sha256':hashlib.sha256(path.read_bytes()).hexdigest()})
    trace=[{'step':1,'event':'Loaded 12 jobs: 4 completed, 6 queued, 2 expired leases'}]
    db.execute("UPDATE jobs SET state='queued' WHERE state='running'");db.commit()
    db.execute("UPDATE jobs SET state='running',attempts=attempts+1 WHERE id='job-05'");db.commit()
    crashed=subprocess.run([sys.executable,str(Path(__file__).resolve()),'--worker',str(dbfile),'queue','job-05']).returncode;assert crashed==73
    trace.append({'step':2,'event':'job-05 result committed; worker exited 73 before acknowledgement'})
    db.close();db=dbconnect(dbfile)
    for id,state,attempts in db.execute('SELECT * FROM jobs ORDER BY id').fetchall():
        if state=='completed':continue
        existing=db.execute('SELECT body FROM results WHERE job_id=?',(id,)).fetchone()
        if existing:trace.append({'step':len(trace)+1,'event':id+': existing result found; acknowledgement restored'})
        else:
            db.execute('INSERT INTO results VALUES(?,?)',(id,'Результат '+id));db.execute('UPDATE jobs SET attempts=attempts+1 WHERE id=?',(id,))
        db.execute("UPDATE jobs SET state='completed' WHERE id=?",(id,));db.commit()
    after=[dict(zip(['id','state','attempts'],x)) for x in db.execute('SELECT * FROM jobs ORDER BY id')]
    assert len(after)==12 and all(x['state']=='completed' for x in after)
    assert db.execute('SELECT count(*) FROM results').fetchone()[0]==12
    assert before[:4]==after[:4]
    for f in media:assert hashlib.sha256((directory/f['path']).read_bytes()).hexdigest()==f['sha256']
    db.close();return {'mode':'local SQLite worker; forced process exit','crashExit':crashed,'before':before,'after':after,'uniqueResults':12,'files':media,'trace':trace,'checks':'PASS'}

if __name__=='__main__':
    if len(sys.argv)>1 and sys.argv[1]=='--worker': worker(sys.argv[2],sys.argv[3],sys.argv[4])
    else:
        mode=sys.argv[1] if len(sys.argv)>1 else 'queue'
        with tempfile.TemporaryDirectory(prefix='aishka-recovery-') as temp:
            result=(automation if mode=='automation' else queue)(Path(temp))
        print(json.dumps(result,ensure_ascii=False,indent=2))