"""Exact-ID customer decisions on invented data; no browser credential claim."""
from concurrent.futures import ThreadPoolExecutor
import csv,io,json,uuid
import pytest
from fastapi.testclient import TestClient
from backend import recovery
from backend.app import create_app
from test_boundaries import app,client,write,job,quote,invoice,portal,ORIGIN
from test_customers import contact,work
from test_workflow import edit,items,decide
from test_fieldwork import captured,draft
from test_photos import reviewed,share,project,pub,record as photo_record
from test_feedback import completed,prepare,respond,publish,BASE,PUBLIC,ACCESS
from test_portal_access import enabled,invited,PASSWORD,response
from test_documents import act,doc
from test_identity import add_staff
from test_recovery import table_rows

ROOT='/api/customer-reviews'

def customer(c,cid):
    r=c.get('/api/customers/'+cid+'/history');assert r.status_code==200,r.text;return r.json()['customer']

def selection(c,source,target,kind='merge',job_ids=None,**fields):
    a=customer(c,source['customer_id']);b=customer(c,target['customer_id'])
    return {'kind':kind,'source_id':a['id'],'target_id':b['id'],'source_version':a['version'],'target_version':b['version'],'job_ids':job_ids or [],'fields':{k:fields.get(k,'target') for k in ('name','email','phone','address')}}

def preview(c,body):
    r=write(c,ROOT+'/transfer-preview',body);assert r.status_code==200,r.text;return r.json()

def confirm(c,plan,key=None,**extra):
    return write(c,ROOT+'/transfers',{**plan['selection'],'preview_hash':plan['preview_hash'],'reason':'Reviewed invented exact customer records','confirmed':True,**extra},key=key)

def complex_history(application,c):
    j,people,clock,tid=captured(c,application)
    photo=reviewed(c,j);assert share(c,j,photo).status_code==200
    gallery=project(c,j,photo);assert pub(c,gallery).status_code==200
    q=draft(c,j,kind='quote').json();act(c,'quote',q,'share')
    p,grant,capture=enabled(application,c,j)
    assert write(p,'/api/portal/'+grant['slug']+'/quotes/'+q['id']+'/accept',response(c,q)).status_code==200
    inv=act(c,'quote',q,'invoice').json();act(c,'invoice',inv,'issue')
    assert write(c,'/api/invoices/'+inv['id']+'/payments',{'amount_pence':10,'note':'Invented receipt'}).status_code==200
    completed(c,j);_,rid,code=prepare(c,j)
    from test_feedback import payload
    assert write(TestClient(application,headers=ORIGIN),'/api/feedback/'+grant['slug']+'/submit',{'code':code,**payload(allow=True)}).status_code==200
    publish(c,rid)
    return {'job':j,'quote':q,'invoice':inv,'portal':p,'portal_id':capture['portal_id'],'setup':capture['code'],'photo':photo,'gallery':gallery,'feedback':rid,'code':code,'timer':tid,'people':people}

def test_exact_history_export_and_matching_contacts_stay_separate(app):
    c=client(app);a=job(c);b=job(c);q=quote(c,a);inv=invoice(c,q)
    assert write(c,'/api/invoices/'+inv['id']+'/payments',{'amount_pence':1200}).status_code==200
    pairs=c.get(ROOT+'/duplicates').json()['pairs'];assert len(pairs)==1 and set(pairs[0]['reasons'])=={'Same name text','Same email text'}
    p=pairs[0];body={'left_id':p['customers'][0]['id'],'right_id':p['customers'][1]['id'],'version':0,'context_hash':p['context_hash'],'state':'keep_separate','reason':'Two different fictional people sharing contact details','confirmed':True}
    assert write(c,ROOT+'/pairs',body).status_code==200
    assert c.get(ROOT+'/duplicates').json()['pairs'][0]['state']=='keep_separate'
    assert len(c.get(ROOT+'/pairs/'+p['id']+'/history').json())==1
    assert c.get('/api/customers/'+a['customer_id']+'/history').json()['totals']['received_pence']==1200
    assert c.get('/api/customers/'+b['customer_id']+'/history').json()['totals']['received_pence']==0
    assert c.get('/api/customers',params={'q':a['customer_id']}).json()[0]['id']==a['customer_id']
    rows=list(csv.DictReader(io.StringIO(c.get('/api/customers/export',params={'q':a['customer_id']}).content.decode('utf-8-sig'))))
    assert len(rows)==1 and rows[0]['customer_id']==a['customer_id'] and rows[0]['recorded_receipts_pence']=='1200'
    forbidden=('portal_hash','password','token','capture','secret','private_media')
    assert not any(x in ','.join(rows[0]) for x in forbidden)

@pytest.mark.parametrize('text',['=2+3',' +SUM(1,2)','-1+2','@SUM(1)','\tformula','\rnext','\nnext','\ufeff=2','\u200b＋2'])
def test_csv_formula_prefixes_are_data_not_formulas(app,text):
    c=client(app);r=write(c,'/api/customers',{'name':'Carys Dêmø','address':'A, B\n"quoted"','phone':text});assert r.status_code==200
    cid=r.json()['id'];assert write(c,'/api/customers/'+cid+'/private-notes',{'version':1,'notes':text},method='put').status_code==200
    response=c.get('/api/customers/export');assert response.status_code==200 and response.headers['cache-control']=='no-store'
    rows=list(csv.DictReader(io.StringIO(response.content.decode('utf-8-sig'))));assert rows[0]['name']=='Carys Dêmø' and rows[0]['address']=='A, B\n"quoted"'
    from backend.customer_history import csv_text
    assert csv_text(text)=="'"+text
    expected={"\tformula":'formula',"\rnext":'next',"\nnext":'next'}.get(text,"'"+text.strip())
    assert rows[0]['phone']==expected and rows[0]['private_notes']==expected

def test_reviewed_merge_preserves_documents_money_sources_and_revokes_permissions(app,tmp_path):
    c=client(app);a=complex_history(app,c);j=a['job'];target=job(c,name='Alex Retained');separate=job(c)
    target_portal,_,_=enabled(app,c,target)
    before=table_rows(app.state.store.root,'oak');values={k:doc(c,k,a[k]) for k in ('quote','invoice')}
    stats=c.get('/api/reports/operations?start=2026-01-01&end=2026-12-31').json()
    p=preview(c,selection(c,j,target,name='source'));assert p['can_confirm'] and len(p['documents'])==2
    key=str(uuid.uuid4());r=confirm(c,p,key);assert r.status_code==200,r.text;assert confirm(c,p,key).json()==r.json()
    after=table_rows(app.state.store.root,'oak')
    for table in ('payments','payment_details','payment_stages','payment_allocations','quote_responses','time_entries','field_intervals','field_time_links','feedback_responses'):
        assert before[table]==after[table],table
    for kind in ('quote','invoice'):
        current=doc(c,kind,a[kind]);assert current['snapshot']==values[kind]['snapshot'] and current['total_pence']==values[kind]['total_pence'] and current['reference']==values[kind]['reference']
        assert current['customer_id']==target['customer_id'] and current['snapshot']['customer']['id']==j['customer_id']
    assert customer(c,j['customer_id'])['state']=='merged' and customer(c,j['customer_id'])['canonical_id']==target['customer_id']
    assert c.get('/api/customers/'+j['customer_id']+'/history').json()['totals']['jobs']==0
    history=c.get('/api/customers/'+target['customer_id']+'/history').json();assert history['totals']['jobs']==2 and history['totals']['received_pence']==10
    assert len(c.get('/api/customers/'+j['customer_id']+'/history').json()['historical_documents'])==2
    assert c.get('/api/jobs/'+separate['id']).json()['customer_id']==separate['customer_id']
    assert a['portal'].get('/api/portal/oak-and-vale/me').status_code==401 and target_portal.get('/api/portal/oak-and-vale/me').status_code==401
    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
    assert write(c,'/api/customers/'+j['customer_id']+'/portal-access/invite',{'version':customer(c,j['customer_id'])['version']}).status_code==409
    assert write(c,'/api/customers/'+j['customer_id']+'/portal').status_code==409
    assert c.get(BASE+a['feedback']).json()['request']['state']=='revoked' and c.get(PUBLIC).json()['count']==0
    assert write(TestClient(app,headers=ORIGIN),ACCESS+'access',{'code':a['code']}).status_code==401
    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 c.get('/api/jobs/'+j['id']+'/workflow-history').json()[-1]['snapshot']['metadata']['followup_excluded']
    new_stats=c.get('/api/reports/operations?start=2026-01-01&end=2026-12-31').json()
    assert new_stats['period_counts']==stats['period_counts'] and new_stats['period_money']==stats['period_money']
    new_portal,_,_=enabled(app,c,target)
    assert len(new_portal.get('/api/portal/oak-and-vale/me').json()['jobs'])==2
    assert new_portal.get('/api/portal/oak-and-vale/jobs/'+j['id']+'/photos').json()==[]
    assert new_portal.get('/api/portal/oak-and-vale/feedback').json()==[]
    saved=recovery.backup(app.state.store.root,tmp_path/'merged.zip');assert recovery.verify(saved['archive'],saved['sha256'])['valid']
    reloaded=create_app(app.state.store.root);assert customer(client(reloaded),j['customer_id'])['state']=='merged'

def test_selected_job_transfer_keeps_source_other_jobs_and_closed_followup(app):
    c=client(app);a=job(c);target=job(c);source=customer(c,a['customer_id']);remaining=work(c,source).json()
    edit(c,a,next_action='Call about exact shelf spacing');item=next(r for r in items(c,'next_action') if r['job_id']==a['id']);decide(c,item,'dismiss')
    plan=preview(c,selection(c,a,target,kind='jobs',job_ids=[a['id']]));r=confirm(c,plan);assert r.status_code==200,r.text
    assert customer(c,a['customer_id'])['state']=='active' and c.get('/api/jobs/'+remaining['id']).json()['customer_id']==a['customer_id']
    edit(c,a,followup_excluded=False,notes='Private edit after explicit review')
    queue=c.get('/api/opportunities').json()['items'];old=next(v for v in queue if v['id']==item['id']);assert old['state']=='dismissed' and not old['eligible']
    assert not any(v['eligible'] and v['job_id']==a['id'] for v in queue)
    edit(c,a,next_action='Genuinely new next step after review');assert any(v['eligible'] and v['job_id']==a['id'] for v in c.get('/api/opportunities').json()['items'])

def test_stale_pair_and_transfer_reviews_preserve_all_records(app):
    c=client(app);a=job(c);b=job(c);p=preview(c,selection(c,a,b));pair=c.get(ROOT+'/duplicates').json()['pairs'][0]
    edit(c,a,notes='Saved after preview')
    before=table_rows(app.state.store.root,'oak');assert confirm(c,p).status_code==409;assert table_rows(app.state.store.root,'oak')==before
    cust=customer(c,a['customer_id']);assert write(c,'/api/customers/'+cust['id'],{'version':cust['version'],'name':'Corrected Demo','email':cust['email']},method='put').status_code==200
    assert write(c,ROOT+'/pairs',{'left_id':pair['customers'][0]['id'],'right_id':pair['customers'][1]['id'],'version':0,'context_hash':pair['context_hash'],'state':'keep_separate','confirmed':True,'reason':'Reviewed separate records'}).status_code==409

def test_concurrent_transfer_has_one_outcome_and_atomic_failure(app,monkeypatch):
    c=client(app);a=job(c);b=job(c);p=preview(c,selection(c,a,b));clients=[client(app),client(app)]
    with ThreadPoolExecutor(2) as pool:responses=list(pool.map(lambda who:confirm(who,p),clients))
    assert sorted(r.status_code for r in responses)==[200,409]
    assert len(table_rows(app.state.store.root,'oak')['customer_transfers'])==1
    x=job(c);y=job(c);p=preview(c,selection(c,x,y));before=table_rows(app.state.store.root,'oak')
    from backend import customer_review
    original=customer_review.record
    def broken(*args,**kwargs):
        original(*args,**kwargs)
        if kwargs.get('advance'):raise RuntimeError('Injected local transaction failure')
    monkeypatch.setattr(customer_review,'record',broken)
    with pytest.raises(RuntimeError):confirm(c,p)
    assert table_rows(app.state.store.root,'oak')==before

def test_owner_business_and_exact_id_boundaries_are_independent_of_shape(app):
    c=client(app);a=job(c);b=job(c);body=selection(c,a,b);p=preview(c,body);staff,_=add_staff(app,c);portal_client,_=portal(app,c,a)
    for other,status in ((staff,403),(portal_client,401),(TestClient(app,headers=ORIGIN),401),(client(app,1),404)):
        assert write(other,ROOT+'/transfer-preview',body).status_code==status
        assert confirm(other,p).status_code==status
    foreign=job(client(app,1));mixed={**body,'target_id':foreign['customer_id']};assert write(c,ROOT+'/transfer-preview',mixed).status_code==404
    assert write(c,ROOT+'/transfer-preview',{**body,'kind':'jobs','job_ids':[foreign['id']]}).status_code==404
    assert staff.get('/api/customers/export').status_code==403 and portal_client.get('/api/customers/export').status_code==401

@pytest.mark.parametrize('blocker',['timer','recurrence','booking','unreviewed_log'])
def test_active_work_blocks_transfer_without_changing_any_association(app,blocker):
    from test_fieldwork import setup,start,base
    from test_recurrence import definition
    from test_scheduling import booking
    c=client(app);j,people,clock=setup(c,app);target=job(c)
    if blocker=='timer':start(c,j,people)
    elif blocker=='recurrence':definition(c,app,j,people[0])
    elif blocker=='booking':booking(c,j,people[0],start_date='2099-01-01',end_date='2099-01-01')
    else:assert write(c,base(j)+'/logs',{'kind':'note','content':'Unreviewed staff observation','work_date':'2026-10-28'}).status_code==200
    p=preview(c,selection(c,j,target));assert not p['can_confirm'] and p['errors']
    before=table_rows(app.state.store.root,'oak');assert confirm(c,p).status_code==409;assert table_rows(app.state.store.root,'oak')==before

def test_customer_private_notes_archive_and_restore_do_not_invent_work_or_permissions(app):
    c=client(app);j,rid,code=prepare(c);respond(c,app,rid,code);publish(c,rid);cust=customer(c,j['customer_id']);p,_=portal(app,c,j)
    assert write(c,'/api/customers/'+cust['id']+'/private-notes',{'version':cust['record_version'],'notes':'Private customer preference'},method='put').status_code==200
    assert c.get(PUBLIC).json()['count']==1 and p.get('/api/portal/oak-and-vale/me').status_code==200
    payload={'version':cust['version'],'state':'archived','confirmed':True,'reason':'Completed relationship retained in history'}
    r=write(c,'/api/customers/'+cust['id']+'/record-state',payload);assert r.status_code==200,r.text
    assert c.get('/api/customers').json()==[] and p.get('/api/portal/oak-and-vale/me').status_code==401
    assert work(c,r.json()).status_code==409
    payload.update(version=r.json()['version'],state='active');assert write(c,'/api/customers/'+cust['id']+'/record-state',payload).status_code==200
    assert c.get(PUBLIC).json()['count']==0 and p.get('/api/portal/oak-and-vale/me').status_code==401

def test_completion_backfill_requires_evidence_exact_links_and_all_rows_valid(app):
    c=client(app);a=job(c);b=job(c);i=invoice(c,quote(c,a));other=invoice(c,quote(c,b));completed(c,a);completed(c,b);edit(c,a,completed_date='');edit(c,b,completed_date='')
    def row(j,inv,date):return {'customer_id':j['customer_id'],'customer_version':customer(c,j['customer_id'])['version'],'job_id':j['id'],'job_version':c.get('/api/jobs/'+j['id']).json()['version'],'invoice_id':inv['id'],'completed_date':date,'evidence_note':'Owner reviewed fictional work completion note'}
    body={'rows':[row(a,i,'2026-09-01'),row(b,other,'2099-09-01')]};p=write(c,ROOT+'/completion-preview',body);assert p.status_code==200 and not p.json()['can_confirm']
    before=table_rows(app.state.store.root,'oak');assert write(c,ROOT+'/completions',{**body,'preview_hash':p.json()['preview_hash'],'confirmed':True,'reason':'Both evidenced completion dates reviewed'}).status_code==409;assert table_rows(app.state.store.root,'oak')==before
    body['rows'][1]['completed_date']='2026-09-02';p=write(c,ROOT+'/completion-preview',body).json();assert p['can_confirm'];key=str(uuid.uuid4());confirm_body={**body,'preview_hash':p['preview_hash'],'confirmed':True,'reason':'Both evidenced completion dates reviewed'}
    r=write(c,ROOT+'/completions',confirm_body,key=key);assert r.status_code==200,r.text;assert write(c,ROOT+'/completions',confirm_body,key=key).json()==r.json()
    after=table_rows(app.state.store.root,'oak');assert before['invoices']==after['invoices'] and before['payments']==after['payments'] and len(before['jobs'])==len(after['jobs'])
    assert [c.get('/api/jobs/'+j['id']).json()['workflow']['completed_date'] for j in (a,b)]==['2026-09-01','2026-09-02']
    wrong=row(a,other,'2026-09-03');assert write(c,ROOT+'/completion-preview',{'rows':[wrong]}).status_code==404

def test_retired_repeat_work_stays_historical_after_transfer(app,tmp_path):
    from test_recurrence import definition,apply,preview as repeat_preview,BASE as REPEAT
    c=client(app);d,j,person=definition(c,app);d=apply(c,d,{'action':'state','state':'retired','reason':'Reviewed retirement before identity correction'})
    target=job(c);plan=preview(c,selection(c,j,target));assert plan['can_confirm'];assert confirm(c,plan).status_code==200
    saved=c.get(REPEAT+'/'+d['id']).json();assert saved['customer_id']==j['customer_id'] and saved['state']=='retired'
    r=write(c,REPEAT+'/'+d['id']+'/preview',{'version':saved['version'],'operation':{'action':'state','state':'restore','reason':'Attempt to restart old identity'}})
    assert r.status_code==409 or (r.status_code==200 and not r.json()['can_confirm'])
    backup=recovery.backup(app.state.store.root,tmp_path/'retired.zip');assert recovery.verify(backup['archive'],backup['sha256'])['valid']


def test_transferred_source_draft_requires_explicit_renewed_review(app,tmp_path):
    from test_fieldwork import docact,doc as source_doc,suggested
    c=client(app);j,people,clock,tid=captured(c,app);d=draft(c,j).json();target=job(c)
    edit(c,j,status='accepted')  # Work is no longer in progress before reviewed correction.
    plan=preview(c,selection(c,j,target));assert confirm(c,plan).status_code==200
    assert docact(c,d,'issue').status_code==409
    saved=source_doc(c,d);assert saved['snapshot']['customer']['id']==j['customer_id']
    # Existing draft re-review explicitly replaces stale source reservations.
    sources=c.get('/api/jobs/'+j['id']+'/suggestions',params={'kind':'invoice','document_id':d['id']}).json()
    r=draft(c,j,sources=sources['sources'],target=saved);assert r.status_code==200,r.text
    assert docact(c,d,'issue').status_code==200
    backup=recovery.backup(app.state.store.root,tmp_path/'rereview.zip');assert recovery.verify(backup['archive'],backup['sha256'])['valid']