"""Cancellation fences and exact saved-log correction in disposable API fixtures."""
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
import json, uuid
import pytest
from backend import recovery
from backend.app import create_app
from test_boundaries import app, client, write
from test_fieldwork import setup, base
from test_device_notes import note, access, action
from test_recovery import new_login


def prepared(app):
    c=client(app);j,_,_=setup(c,app);op,p,_=note(c,j)
    return c,j,op,p


def proposal(p,version=1):
    return {'binding':p['binding'],'edit_id':str(uuid.uuid4()),'version':version,'content':'Reviewed private proposed text','work_date':'2026-10-30','reason':'Reviewed exact saved note'}


def test_cancel_never_received_then_delayed_deliver_revise_replay(app):
    c,j,op,p=prepared(app);body={'binding':p['binding']}
    cancelled=action(c,j,op,body,'cancel');assert cancelled.status_code==200
    assert cancelled.json()['state']=='cancelled'
    assert action(c,j,op,body,'cancel').json()==cancelled.json()
    assert action(c,j,op,body,'inspect').json()==cancelled.json()
    assert action(c,j,op,p).status_code==409
    assert action(c,j,op,{**body,'revision':2},'revise').status_code==409
    with app.state.store.tenant('oak') as db:
        assert db.execute('SELECT COUNT(*) FROM field_logs').fetchone()[0]==0
        assert db.execute('SELECT fingerprint FROM operations WHERE key=?',(op,)).fetchone()[0].startswith('device-note-closed-v2:')


@pytest.mark.parametrize('repeat',range(4))
def test_cancel_concurrent_with_send_is_one_terminal_outcome(app,repeat):
    c,j,op,p=prepared(app);other=client(app)
    with ThreadPoolExecutor(2) as pool:
        sent=pool.submit(action,c,j,op,p);cancelled=pool.submit(action,other,j,op,{'binding':p['binding']},'cancel')
        a,b=sent.result(),cancelled.result()
    state=b.json()['state'];assert state in ('cancelled','confirmed')
    with app.state.store.tenant('oak') as db:
        count=db.execute('SELECT COUNT(*) FROM field_logs').fetchone()[0];assert count==int(state=='confirmed')
        assert db.execute('SELECT COUNT(*) FROM field_log_history').fetchone()[0]==count
    assert a.status_code==(200 if state=='confirmed' else 409)
    for _ in range(2):assert action(c,j,op,{'binding':p['binding']},'cancel').json()==b.json()


def test_cancel_racing_revision_never_allows_late_delivery(app):
    c,j,op,p=prepared(app);other=client(app);b={'binding':p['binding']}
    with ThreadPoolExecutor(2) as pool:
        revision=pool.submit(action,c,j,op,{**b,'revision':2},'revise');cancel=pool.submit(action,other,j,op,b,'cancel')
        assert revision.result().status_code in (200,409);assert cancel.result().json()['state']=='cancelled'
    assert action(c,j,op,{**p,'revision':2}).status_code==409


def test_cancel_lost_commit_reply_shows_exact_saved_log_and_keeps_history(app):
    c,j,op,p=prepared(app);saved=action(c,j,op,p).json();b={'binding':p['binding']}
    assert action(c,j,op,b,'cancel').json()==saved
    view=action(c,j,op,b,'saved').json();assert view['log']['id']==saved['id'] and view['log']['content']==p['content']
    assert len(c.get(base(j)).json()['logs'])==1


def test_apply_candidate_uses_ordinary_versioned_edit_and_exact_retry(app):
    c,j,op,p=prepared(app);saved=action(c,j,op,p).json();edit=proposal(p)
    first=action(c,j,op,edit,'apply-correction');assert first.status_code==200,first.text
    assert first.json()['state']=='applied' and first.json()['version']==2 and first.json()['id']==saved['id']
    assert action(c,j,op,edit,'apply-correction').json()==first.json()
    assert action(c,j,op,{'binding':p['binding'],'edit_id':edit['edit_id']},'resolve-correction').json()==first.json()
    view=action(c,j,op,{'binding':p['binding']},'saved').json();assert view['log']['content']==edit['content'] and view['log']['version']==2
    with app.state.store.tenant('oak') as db:
        assert db.execute('SELECT COUNT(*) FROM field_logs').fetchone()[0]==1
        rows=db.execute('SELECT action,reason,snapshot FROM field_log_history ORDER BY id').fetchall()
        assert [r['action'] for r in rows]==['created','edited']
        assert rows[1]['reason']==edit['reason'] and json.loads(rows[0]['snapshot'])['content']==p['content']


def test_cancel_correction_before_delayed_apply_leaves_original_unchanged(app):
    c,j,op,p=prepared(app);action(c,j,op,p);edit=proposal(p)
    b={'binding':p['binding'],'edit_id':edit['edit_id']}
    first=action(c,j,op,b,'resolve-correction');assert first.json()['state']=='cancelled'
    assert action(c,j,op,b,'resolve-correction').json()==first.json()
    assert action(c,j,op,edit,'apply-correction').status_code==409
    assert action(c,j,op,{'binding':p['binding']},'saved').json()['log']['content']==p['content']


def test_stale_or_invalid_correction_can_be_fenced_then_reviewed_again(app):
    c,j,op,p=prepared(app);action(c,j,op,p);edit=proposal(p);edit['reason']=''
    assert action(c,j,op,edit,'apply-correction').status_code==422
    assert action(c,j,op,{'binding':p['binding'],'edit_id':edit['edit_id']},'resolve-correction').json()['state']=='cancelled'
    changed=proposal(p);assert action(c,j,op,changed,'apply-correction').status_code==200
    stale=proposal(p);assert action(c,j,op,stale,'apply-correction').status_code==409
    assert action(c,j,op,{'binding':p['binding'],'edit_id':stale['edit_id']},'resolve-correction').json()['state']=='cancelled'
    reviewed=proposal(p,version=2);assert action(c,j,op,reviewed,'apply-correction').json()['version']==3
    assert len(c.get(base(j)).json()['logs'])==1


def test_saved_photo_references_survive_text_correction(app):
    from test_photos import reviewed
    c,j,op,p=prepared(app);saved=action(c,j,op,p).json();photo=reviewed(c,j)
    old={'kind':'note','content':p['content'],'work_date':p['work_date'],'version':1,'reason':'Attach exact private photo','photo_ids':[photo['id']]}
    assert write(c,base(j)+'/logs/'+saved['id'],old,method='put').status_code==200
    assert action(c,j,op,proposal(p,2),'apply-correction').status_code==200
    view=action(c,j,op,{'binding':p['binding']},'saved').json()['log'];assert view['photo_ids']==[photo['id']]


@pytest.mark.parametrize('loss',['foreign','membership','assignment','closed','transferred'])
def test_new_routes_recheck_scope_before_receipts(app,loss):
    from test_identity import add_staff
    from test_scheduling import booking
    from test_boundaries import job
    owner=client(app);j,people,_=setup(owner,app);booking(owner,j,people[0]);c,_=add_staff(app,owner,resource_id=people[0]['id']);op,p,_=note(c,j);action(c,j,op,p);edit=proposal(p)
    assert action(c,j,op,edit,'apply-correction').status_code==200
    if loss=='foreign':c=owner
    elif loss=='membership':
        with app.state.store.registry() as db:db.execute("UPDATE memberships SET enabled=0 WHERE role='staff'")
    elif loss=='assignment':
        with app.state.store.tenant('oak') as db:db.execute('UPDATE calendar_bookings SET archived=1')
    elif loss=='closed':
        with app.state.store.tenant('oak') as db:db.execute("UPDATE jobs SET status='completed' WHERE id=?",(j['id'],))
    else:
        other=job(owner,name='Other invented customer',email='other@example.test')
        with app.state.store.tenant('oak') as db:db.execute('UPDATE jobs SET customer_id=(SELECT customer_id FROM jobs WHERE id=?) WHERE id=?',(other['id'],j['id']))
    for verb,body in [('cancel',{'binding':p['binding']}),('saved',{'binding':p['binding']}),('apply-correction',edit),('resolve-correction',{'binding':p['binding'],'edit_id':edit['edit_id']})]:
        assert action(c,j,op,body,verb).status_code in (401,403,404),verb


def test_archived_saved_log_denies_edit_but_allows_discarding_pending_attempt(app):
    c,j,op,p=prepared(app);saved=action(c,j,op,p).json();edit=proposal(p)
    assert write(c,base(j)+'/logs/'+saved['id']+'/state/archive',{'version':1,'reason':'Owner archived'}).status_code==200
    assert action(c,j,op,edit,'apply-correction').status_code==403
    assert action(c,j,op,{'binding':p['binding'],'edit_id':edit['edit_id']},'resolve-correction').json()['state']=='cancelled'


def test_cancellation_and_correction_receipts_survive_current_recovery(app,tmp_path):
    c,j,op,p=prepared(app);action(c,j,op,p);edit=proposal(p);action(c,j,op,edit,'apply-correction')
    cancelled=str(uuid.uuid4());action(c,j,cancelled,{'binding':p['binding']},'cancel')
    never=proposal(p,2);action(c,j,op,{'binding':p['binding'],'edit_id':never['edit_id']},'resolve-correction')
    with app.state.store.tenant('oak') as db:before=[tuple(r) for r in db.execute('SELECT * FROM operations ORDER BY key')]
    archive=tmp_path/'lifecycle.zip';b=recovery.backup(app.state.store.root,archive);assert recovery.verify(archive,b['sha256'])['valid']
    target=tmp_path/'restore';recovery.restore(archive,b['sha256'],target,confirm_fresh=True);restored=create_app(target)
    with restored.state.store.tenant('oak') as db:
        assert [tuple(r) for r in db.execute('SELECT * FROM operations ORDER BY key')]==before
        assert db.execute('SELECT COUNT(*) FROM field_logs').fetchone()[0]==1
        assert db.execute('SELECT COUNT(*) FROM field_log_history').fetchone()[0]==2
    assert json.loads((target/'RESTORE-REPORT.json').read_text())['recovery_format']==14
    after,_=new_login(restored,target,'oak');assert action(after,j,op,{'binding':p['binding']},'saved').status_code==403
    second=tmp_path/'second.zip';b=recovery.backup(target,second);assert recovery.verify(second,b['sha256'])['valid']