"""Repeated exact corrections, immutable provenance, stale/atomic boundaries and v12 recovery."""
from concurrent.futures import ThreadPoolExecutor
import copy, hashlib, json, shutil, sqlite3, uuid, zipfile
from pathlib import Path
import pytest
from fastapi.testclient import TestClient
from backend import recovery
from backend.app import create_app
from backend.customer_recovery import validate
from test_boundaries import app, client, write, job, ORIGIN
from test_customer_reviews import complex_history, customer, selection, preview, confirm, ROOT
from test_customers import work
from test_recovery import table_rows, new_login, rewrite_archive
from test_documents import doc
from test_device_notes import access, action
from test_fieldwork import setup
from test_recurrence import definition, apply as recurrence_action
from test_workflow import edit, items, decide


def move(c,j,target,ids=None,key=None):
    current=c.get('/api/jobs/'+j['id']).json()
    p=preview(c,selection(c,current,target,kind='jobs',job_ids=ids or [j['id']]))
    assert p['can_confirm'],p['errors']
    r=confirm(c,p,key);assert r.status_code==200,r.text
    return p,r.json()


def check(application):
    with application.state.store.tenant('oak') as db:
        def require(ok,message):assert ok,message
        return validate(db,require)


@pytest.mark.parametrize('route',[(1,2),(1,0),(1,2,0),(1,0,1,0)])
def test_ordered_moves_preserve_sealed_money_operational_and_access_history(app,tmp_path,route):
    c=client(app);a=complex_history(app,c);j=a['job'];targets=[j,job(c,name='Blake Fictional'),job(c,name='Carys Fictional')]
    untouched=work(c,customer(c,j['customer_id'])).json();before=table_rows(app.state.store.root,'oak')
    docs={k:doc(c,k,a[k]) for k in ('quote','invoice')};report=c.get('/api/reports/operations?start=2026-01-01&end=2026-12-31').json()
    moves=[]
    for index in route:
        p,r=move(c,j,targets[index]);moves.append(r)
        assert len(p['ownership_chains'][j['id']])==len(moves)-1
        check(app)
        assert a['portal'].get('/api/portal/oak-and-vale/me').status_code==401
        from test_portal_access import PASSWORD
        assert a['portal'].post('/api/portal/oak-and-vale/access/login',json={'portal_id':a['portal_id'],'password':PASSWORD}).status_code==401
        assert a['portal'].post('/api/portal/oak-and-vale/access/complete',json={'code':a['setup'],'password':PASSWORD}).status_code==401
        from test_photos import record as photo_record
        photo=photo_record(c,j,a['photo']);assert photo['customer_share'] is None and not photo['reviewed']
        assert c.get('/api/gallery/'+a['gallery']['id']).json()['published'] is None
        assert write(TestClient(app,headers=ORIGIN),'/api/feedback/oak-and-vale/access',{'code':a['code']}).status_code==401
        assert c.get('/api/public/oak-and-vale/testimonials').json()['count']==0
    after=table_rows(app.state.store.root,'oak')
    for name in ('payments','payment_details','payment_stages','payment_allocations','quote_responses','time_entries','field_intervals','field_time_links','feedback_responses'):
        assert before[name]==after[name],name
    for kind in docs:
        d=doc(c,kind,a[kind]);assert d['snapshot']==docs[kind]['snapshot'] and d['reference']==docs[kind]['reference'] and d['total_pence']==docs[kind]['total_pence']
    assert c.get('/api/jobs/'+untouched['id']).json()['customer_id']==untouched['customer_id']
    h=c.get('/api/customers/'+targets[route[-1]]['customer_id']+'/history').json()
    chain=next(h for h in h['ownership_history'] if h['job_id']==j['id'])
    assert [m['sequence'] for m in chain['moves']]==list(range(1,len(route)+1))
    assert [m['transfer_id'] for m in chain['moves']]==[m['id'] for m in moves]
    assert chain['observed_original_customer_id']==j['customer_id'] and chain['current_customer_id']==targets[route[-1]]['customer_id']
    updated_report=c.get('/api/reports/operations?start=2026-01-01&end=2026-12-31').json()
    # Financial totals are independent of grouping by current customer.
    assert report['period_money']==updated_report['period_money'] and report['period_counts']==updated_report['period_counts']
    b=recovery.backup(app.state.store.root,tmp_path/'chain.zip');assert recovery.verify(b['archive'],b['sha256'])['valid']
    restored=tmp_path/'recovered';recovery.restore(b['archive'],b['sha256'],restored,confirm_fresh=True)
    restored_app=create_app(restored);check(restored_app)
    got=table_rows(restored,'oak')
    for name in ('customer_transfer_jobs','customer_transfers','quotes','invoices','document_events','payments','source_history','field_history','field_log_history','photo_history'):
        assert got[name]==after[name],name
    second=recovery.backup(restored,tmp_path/'second.zip');assert recovery.verify(second['archive'],second['sha256'])['valid']


def test_return_keeps_old_note_bindings_receipts_and_retired_repeat_rule_closed(app):
    c=client(app);j,people,_=setup(c,app);edit(c,j,status='accepted');b=job(c);a=access(c,j);op=str(uuid.uuid4())
    assert action(c,j,op,{'binding':a['binding']},'cancel').status_code==200
    d,_,_=definition(c,app,j,people[0]);d=recurrence_action(c,d,{'action':'state','state':'retired','reason':'Close invented repeat definition before correction'})
    move(c,j,b);middle=access(c,j,a['binding']['device_id']);move(c,j,j);returned=access(c,j,a['binding']['device_id'])
    assert len({a['key'],middle['key'],returned['key']})==3
    for old in (a,middle):
        for verb in ('inspect','cancel','saved'):
            assert action(c,j,op,{'binding':old['binding']},verb).status_code==403
    # Same operation cannot be claimed under a new scope either.
    assert action(c,j,op,{'binding':returned['binding']},'inspect').status_code==409
    r=write(c,'/api/recurring/'+d['id']+'/preview',{'version':d['version'],'operation':{'action':'state','state':'restore','reason':'Try restoring old definition after return'}})
    assert r.status_code==409,r.text
    check(app)


@pytest.mark.parametrize('change',['new_job','note','customer','portal','active_work','one_wrong_job'])
def test_stale_or_invalid_batch_changes_nothing_and_retains_preview(app,change):
    c=client(app);a=job(c);b=job(c);second=work(c,customer(c,a['customer_id'])).json()
    move(c,a,b,ids=[a['id'],second['id']]);edit(c,a,status='accepted');current=c.get('/api/jobs/'+a['id']).json()
    p=preview(c,selection(c,current,a,kind='jobs',job_ids=[a['id'],second['id']]))
    if change=='new_job':work(c,customer(c,b['customer_id']))
    elif change=='note':
        # A new association makes the saved dependency inventory stale.
        from test_device_notes import note
        op,body,_=note(c,current);assert action(c,current,op,body).status_code==200
    elif change=='customer':
        row=customer(c,a['customer_id']);write(c,'/api/customers/'+row['id']+'/private-notes',{'version':row['record_version'],'notes':'New private customer fact'},method='put')
    elif change=='portal':write(c,'/api/customers/'+b['customer_id']+'/portal-access/invite',{'version':customer(c,b['customer_id'])['version']})
    elif change=='active_work':
        with app.state.store.tenant('oak') as db:db.execute("UPDATE jobs SET status='in_progress' WHERE id=?",(second['id'],))
    else:p['selection']['job_ids'].append(job(c)['id'])
    before=table_rows(app.state.store.root,'oak');r=confirm(c,p);assert r.status_code in (404,409),r.text
    assert before==table_rows(app.state.store.root,'oak')


def test_concurrent_and_lost_reply_use_one_durable_operation_even_after_cache_pruned(app):
    c=client(app);a=job(c);b=job(c);move(c,a,b)
    p=preview(c,selection(c,c.get('/api/jobs/'+a['id']).json(),a,kind='jobs',job_ids=[a['id']]))
    key=str(uuid.uuid4());other=client(app)
    with ThreadPoolExecutor(2) as pool:responses=list(pool.map(lambda cli:confirm(cli,p,key),[c,other]))
    assert all(r.status_code==200 for r in responses);assert responses[0].json()==responses[1].json()
    with app.state.store.tenant('oak') as db:db.execute('DELETE FROM operations WHERE key=?',(key,))
    before=table_rows(app.state.store.root,'oak');assert confirm(c,p,key).json()==responses[0].json()
    after=table_rows(app.state.store.root,'oak')
    assert all(before[t]==after[t] for t in before if t!='operations')
    assert confirm(c,p,key,reason='A conflicting reason with the same operation').status_code==409
    check(app)


def test_alias_bearing_canonical_can_correct_job_but_alias_cannot_receive_and_merge_stays_bounded(app):
    c=client(app);a=job(c);b=job(c);d=job(c);r=confirm(c,preview(c,selection(c,a,b)));assert r.status_code==200
    move(c,a,d);move(c,a,b);check(app)
    current=c.get('/api/jobs/'+a['id']).json()
    assert write(c,ROOT+'/transfer-preview',selection(c,current,a,kind='jobs',job_ids=[a['id']])).status_code==409
    assert not preview(c,selection(c,b,d))['can_confirm']


def legacy_restored(tmp_path):
    fixture=Path(__file__).parent/'fixtures/v0.23-customer-moves-format12.zip';meta=json.loads(fixture.with_suffix('.json').read_text())
    assert hashlib.sha256(fixture.read_bytes()).hexdigest()==meta['fixture_sha256']
    root=tmp_path/'restored';recovery.restore(fixture,meta['fixture_sha256'],root,confirm_fresh=True)
    application=create_app(root);c,_=new_login(application,root,'oak')
    return application,c,meta,fixture


def test_genuine_v23_format12_upgrade_then_repeated_correction_and_second_backup(tmp_path):
    application,c,meta,fixture=legacy_restored(tmp_path);ids=list(meta['jobs'].values())[:3]
    with zipfile.ZipFile(fixture) as archive:
        old=tmp_path/'old.sqlite3';old.write_bytes(archive.read('businesses/oak/business.sqlite3'))
    with sqlite3.connect(old) as db:
        original=db.execute('SELECT * FROM customer_transfer_jobs ORDER BY job_id').fetchall()
        original_reviews=db.execute('SELECT * FROM customer_transfers ORDER BY id').fetchall()
    with application.state.store.tenant('oak') as db:
        upgraded=[tuple(r) for r in db.execute('SELECT * FROM customer_transfer_jobs ORDER BY job_id')]
        assert [r[:6] for r in upgraded]==original and all(r[6:]==(1,None) for r in upgraded)
        assert [tuple(r) for r in db.execute('SELECT * FROM customer_transfers ORDER BY id')]==original_reviews
    target=job(c,name='Carys Later Review');j=c.get('/api/jobs/'+ids[0]).json()
    move(c,j,target,ids);original_target={'customer_id':meta['customers']['original']};move(c,j,original_target,ids)
    check(application)
    after=table_rows(application.state.store.root,'oak')
    assert next(j for j in after['jobs'] if j['id']==meta['jobs']['unselected'])['customer_id']==meta['customers']['original']
    assert not table_rows(application.state.store.root,'harbour')['jobs']
    b=recovery.backup(application.state.store.root,tmp_path/'upgraded.zip');assert recovery.verify(b['archive'],b['sha256'])['valid']
    with zipfile.ZipFile(b['archive']) as archive:
        metadata=json.loads(archive.read('manifest.json'));assert (metadata['version'],metadata['prototype_version'])==(14,'0.25.0')
    again=tmp_path/'again';recovery.restore(b['archive'],b['sha256'],again,confirm_fresh=True)
    for table in ('customer_transfers','customer_transfer_jobs','enquiry_intakes','enquiry_photo_sources','recurring_contracts','recurrence_occurrences','quotes','invoices','payments'):
        assert after[table]==table_rows(again,'oak')[table],table
    second=recovery.backup(again,tmp_path/'second.zip');assert recovery.verify(second['archive'],second['sha256'])['valid']


@pytest.mark.parametrize('tamper',['missing','branched','out_of_order','previous','job_owner','plan_chain','document_event','field_event','alias_cycle'])
def test_recovery_rejects_incomplete_or_forged_chains_before_creating_target(app,tmp_path,tamper):
    c=client(app);a=complex_history(app,c);j=a['job'];b=job(c);d=job(c)
    move(c,j,b);move(c,j,d);move(c,j,j)
    source=tmp_path/'source.zip';recovery.backup(app.state.store.root,source)
    def mutate(files,meta):
        name='businesses/oak/business.sqlite3';p=tmp_path/'tamper.sqlite3';p.write_bytes(files[name])
        with sqlite3.connect(p) as db:
            if tamper=='missing':db.execute('DELETE FROM customer_transfer_jobs WHERE sequence=2')
            elif tamper=='branched':db.execute('UPDATE customer_transfer_jobs SET from_customer_id=? WHERE sequence=2',(j['customer_id'],))
            elif tamper=='out_of_order':db.execute('UPDATE customer_transfer_jobs SET sequence=10 WHERE sequence=2')
            elif tamper=='previous':db.execute('UPDATE customer_transfer_jobs SET previous_transfer_id=NULL WHERE sequence=3')
            elif tamper=='job_owner':db.execute('UPDATE jobs SET customer_id=? WHERE id=?',(b['customer_id'],j['id']))
            elif tamper=='plan_chain':
                tid=db.execute('SELECT transfer_id FROM customer_transfer_jobs WHERE sequence=3').fetchone()[0]
                raw=json.loads(db.execute('SELECT plan FROM customer_transfers WHERE id=?',(tid,)).fetchone()[0]);raw['ownership_chains'][j['id']].pop(0)
                from backend.customer_review import packed
                from backend.store import digest
                raw['preview_hash']=digest(packed({k:v for k,v in raw.items() if k!='preview_hash'}));db.execute('UPDATE customer_transfers SET plan=?,preview_hash=? WHERE id=?',(json.dumps(raw),raw['preview_hash'],tid))
            elif tamper=='document_event':db.execute("DELETE FROM document_events WHERE id=(SELECT MAX(id) FROM document_events WHERE action='customer.transferred')")
            elif tamper=='field_event':db.execute("DELETE FROM field_history WHERE id=(SELECT MAX(id) FROM field_history WHERE action='customer.transferred')")
            else:db.execute("UPDATE customer_records SET state='merged',canonical_id=customer_id")
        files[name]=p.read_bytes();meta['files'][name].update(bytes=len(files[name]),sha256=hashlib.sha256(files[name]).hexdigest())
    bad=tmp_path/'bad.zip';sha=rewrite_archive(source,bad,mutate);destination=tmp_path/'must-not-exist'
    with pytest.raises((recovery.RecoveryError,ValueError,KeyError,AssertionError)):recovery.restore(bad,sha,destination,confirm_fresh=True)
    assert not destination.exists()


def test_return_does_not_reopen_old_followup_or_permission_when_explicitly_unexcluded(app):
    c=client(app);a=job(c);b=job(c)
    edit(c,a,next_action='Call about invented original shelf size');old=next(i for i in items(c,'next_action') if i['job_id']==a['id']);decide(c,old,'dismiss')
    move(c,a,b);move(c,a,a);edit(c,a,followup_excluded=False)
    queue=items(c);assert next(i for i in queue if i['id']==old['id'])['eligible'] is False
    assert not any(i['eligible'] and i['job_id']==a['id'] for i in queue)
    edit(c,a,next_action='A new individually reviewed step after correction');fresh=next(i for i in items(c) if i['eligible'] and i['job_id']==a['id']);assert fresh['id']!=old['id']
    decide(c,fresh,'prepare');move(c,a,b);move(c,a,a);edit(c,a,followup_excluded=False)
    assert all(not i['eligible'] for i in items(c) if i['job_id']==a['id'])
    check(app)


def test_second_correction_batch_fault_rolls_back_all_history_then_same_attempt_succeeds(app,monkeypatch):
    import backend.customer_review as review
    c=client(app);a=job(c);b=job(c);second=work(c,customer(c,a['customer_id'])).json();ids=[a['id'],second['id']]
    move(c,a,b,ids);p=preview(c,selection(c,c.get('/api/jobs/'+a['id']).json(),a,kind='jobs',job_ids=ids));key=str(uuid.uuid4())
    original=review.record
    def broken(db,cid,*args,**kwargs):
        original(db,cid,*args,**kwargs)
        if cid==a['customer_id']:raise RuntimeError('Injected late batch failure')
    before=table_rows(app.state.store.root,'oak');monkeypatch.setattr(review,'record',broken)
    with pytest.raises(RuntimeError,match='Injected'):confirm(c,p,key)
    assert before==table_rows(app.state.store.root,'oak')
    monkeypatch.setattr(review,'record',original);assert confirm(c,p,key).status_code==200;check(app)


@pytest.mark.parametrize('scope',['foreign','revoked','role'])
def test_current_authority_is_required_for_new_or_cached_return_correction(app,scope):
    c=client(app);a=job(c);b=job(c);move(c,a,b)
    p=preview(c,selection(c,c.get('/api/jobs/'+a['id']).json(),a,kind='jobs',job_ids=[a['id']]));key=str(uuid.uuid4());assert confirm(c,p,key).status_code==200
    if scope=='foreign':c=client(app,1)
    else:
        with app.state.store.registry() as db:
            if scope=='revoked':db.execute("UPDATE memberships SET enabled=0 WHERE user_id='oak'")
            else:db.execute("UPDATE memberships SET role='staff' WHERE user_id='oak'")
    before=table_rows(app.state.store.root,'oak');assert confirm(c,p,key).status_code in (401,403,404);assert before==table_rows(app.state.store.root,'oak')


def test_active_work_in_one_selected_job_blocks_entire_fresh_return_preview(app):
    c=client(app);a=job(c);b=job(c);second=work(c,customer(c,a['customer_id'])).json();move(c,a,b,[a['id'],second['id']])
    edit(c,second,status='in_progress');p=preview(c,selection(c,c.get('/api/jobs/'+a['id']).json(),a,kind='jobs',job_ids=[a['id'],second['id']]))
    assert not p['can_confirm'] and any('in progress' in e for e in p['errors'])
    before=table_rows(app.state.store.root,'oak');assert confirm(c,p).status_code==409;assert before==table_rows(app.state.store.root,'oak')


def test_new_portal_reset_token_stales_exact_preview(app):
    from backend.portal_access import token as make_token
    from test_portal_access import enabled
    c=client(app);a=job(c);b=job(c);move(c,a,b);current=c.get('/api/jobs/'+a['id']).json()
    enabled(app,c,current);p=preview(c,selection(c,current,a,kind='jobs',job_ids=[a['id']]))
    # Reset tokens carry no permission to change this job, but change access facts.
    with app.state.store.tenant('oak') as db:
        account=dict(db.execute('SELECT * FROM portal_accounts WHERE customer_id=?',(b['customer_id'],)).fetchone())
        make_token(db,account['customer_id'],'reset',account['version'],1800)
    before=table_rows(app.state.store.root,'oak');assert confirm(c,p).status_code==409;assert before==table_rows(app.state.store.root,'oak')


def test_invalid_legacy_transfer_is_rejected_before_upgrade_or_restore(tmp_path):
    fixture=Path(__file__).parent/'fixtures/v0.23-customer-moves-format12.zip'
    def corrupt(files,meta):
        name='businesses/oak/business.sqlite3';p=tmp_path/'old.sqlite3';p.write_bytes(files[name])
        with sqlite3.connect(p) as db:db.execute('DELETE FROM customer_transfer_jobs WHERE job_id=(SELECT job_id FROM customer_transfer_jobs LIMIT 1)')
        files[name]=p.read_bytes();meta['files'][name].update(bytes=len(files[name]),sha256=hashlib.sha256(files[name]).hexdigest())
    bad=tmp_path/'bad-old.zip';sha=rewrite_archive(fixture,bad,corrupt);target=tmp_path/'blocked'
    with pytest.raises((recovery.RecoveryError,ValueError)):recovery.restore(bad,sha,target,confirm_fresh=True)
    assert not target.exists()