"""Invented operational records only; no scheduler, provider or message delivery."""
from concurrent.futures import ThreadPoolExecutor
from datetime import date,timedelta
import json,uuid
import pytest
from fastapi.testclient import TestClient
from backend import workflow
from backend.workflow_models import Metadata
from test_boundaries import app,client,write,job,quote,invoice,portal,ORIGIN
from test_documents import act,doc,schedule,draft_invoice,edit_payload
from test_feedback import prepare,respond,publish,action,BASE,ACCESS,PUBLIC,consent
from test_identity import add_staff
from test_recurrence import definition,apply
from test_scheduling import booking
from test_workforce import person,entry
from test_materials import material

def current(c,j):return c.get('/api/jobs/'+j['id']).json()
def edit_body(c,j,**changes):
    r=current(c,j);body={k:r[k] for k in ('version','status','notes','scheduled_date')};body['metadata']=r['workflow'];body.update({k:v for k,v in changes.items() if k not in Metadata.model_fields});body['metadata'].update({k:v for k,v in changes.items() if k in Metadata.model_fields});return body
def edit(c,j,**changes):
    r=write(c,'/api/jobs/'+j['id']+'/workflow',edit_body(c,j,**changes),method='put');assert r.status_code==200,r.text;return current(c,j)
def config(c,**changes):
    r=write(c,'/api/opportunity-settings',{**c.get('/api/opportunity-settings').json(),**changes},method='put');assert r.status_code==200,r.text
def items(c,kind=None):
    r=c.get('/api/opportunities');assert r.status_code==200,r.text;return [i for i in r.json()['items'] if kind is None or i['kind']==kind]
def review_body(i,action='prepare',**extra):return {'version':i['version'],'context_hash':i['context_hash'],'action':action,'confirmed':True,'draft':'A reviewed local draft only' if action=='prepare' else '','note':'Reviewed fictional decision',**extra}
def decide(c,i,action='prepare',**extra):
    r=write(c,'/api/opportunities/'+i['id']+'/actions',review_body(i,action,**extra));assert r.status_code==200,r.text;return c.get('/api/opportunities/'+i['id']).json()['item']

def test_exact_pipeline_filters_counts_private_metadata_and_no_financial_effects(app):
    c=client(app);a=job(c);b=job(c);q=quote(c,a);before=doc(c,'quote',q)
    edit(c,a,priority='urgent',tags=['Repair','Call'],next_action='Call about shelf height',next_action_date=workflow.today(),notes='private orange');edit(c,b,priority='low',tags=['later'])
    data=c.get('/api/pipeline',params={'q':'orange','priority':'urgent','tag':'call','stage':'new','due':'due'}).json()
    assert [j['id'] for j in data['jobs']]==[a['id']] and data['matching']==1 and data['stored']==2 and data['stages']['new']==1
    assert data['jobs'][0]['workflow']['tags']==['call','repair']
    assert c.get('/api/pipeline',params={'q':'missing','due':'due'}).json()['matching']==0
    assert c.get('/api/pipeline',params={'q':b['customer_id']}).json()['jobs'][0]['id']==b['id']
    p,_=portal(app,c,a);public=p.get('/api/portal/oak-and-vale/me').text
    for private in ('orange','urgent','next_action','workflow','repair'):assert private not in public
    assert doc(c,'quote',q)==before and c.get('/api/workspace').json()['invoices']==[]

def test_job_save_idempotency_race_and_stale_rejection(app):
    c=client(app);j=job(c);body=edit_body(c,j,priority='urgent');url='/api/jobs/'+j['id']+'/workflow';key=str(uuid.uuid4())
    first=write(c,url,body,method='put',key=key);assert first.status_code==200
    assert write(c,url,body,method='put',key=key).json()==first.json();assert write(c,url,body,method='put').status_code==409
    body=edit_body(c,j,notes='Concurrent owner edit');clients=[client(app),client(app)]
    with ThreadPoolExecutor(2) as pool:results=list(pool.map(lambda cc:write(cc,url,body,method='put'),clients))
    assert sorted(r.status_code for r in results)==[200,409];assert len(c.get('/api/jobs/'+j['id']+'/workflow-history').json())==3

@pytest.mark.parametrize('change',[{'priority':'highest'},{'tags':['a','A']},{'tags':['bad!']},{'next_action_date':'2026-10-28'},{'completed_date':'2026-02-30'},{'last_contact_date':'2099-01-01'},{'postcode_group':'unknown'},{'archived':1},{'version':True},{'next_action':'  ','next_action_date':'2026-10-28'},{'scheduled_date':'20261028'}])
def test_invalid_metadata_does_not_write(app,change):
    c=client(app);j=job(c);r=write(c,'/api/jobs/'+j['id']+'/workflow',edit_body(c,j,**change),method='put');assert r.status_code==422,r.text;assert current(c,j)['version']==1
    if 'scheduled_date' in change:
        assert write(c,'/api/jobs/'+j['id'],{'version':1,'status':'scheduled',**change},method='put').status_code==422

def test_archive_guards_future_booking_repeat_contract_draft_billing_and_reopen(app):
    c=client(app);j=job(c);p=person(c);booking(c,j,p,start_date='2099-01-01',end_date='2099-01-01');edit(c,j,status='completed')
    assert write(c,'/api/jobs/'+j['id']+'/workflow',edit_body(c,j,archived=True),method='put').status_code==409
    d,j2,_=definition(c,app);edit(c,j2,status='completed');assert write(c,'/api/jobs/'+j2['id']+'/workflow',edit_body(c,j2,archived=True),method='put').status_code==409
    j3,q,i=draft_invoice(c);edit(c,j3,status='completed');assert write(c,'/api/jobs/'+j3['id']+'/workflow',edit_body(c,j3,archived=True),method='put').status_code==409
    j4=job(c);edit(c,j4,status='cancelled',archived=True)
    assert c.get('/api/pipeline',params={'archive':'archived'}).json()['jobs'][0]['id']==j4['id']
    assert write(c,'/api/jobs/'+j4['id']+'/workflow',edit_body(c,j4,status='new'),method='put').status_code==409
    edit(c,j4,status='new',archived=False)

def test_quote_suggestions_current_shared_age_last_contact_revision_and_status(app,monkeypatch):
    c=client(app);j=job(c);q=quote(c,j);assert not items(c,'quote');act(c,'quote',q,'share')
    with app.state.store.tenant('oak') as db:db.execute("UPDATE quote_links SET shared_at='2026-10-01T23:30:00+00:00'")
    monkeypatch.setattr(workflow,'today',lambda:'2026-10-09')
    one=items(c,'quote');assert len(one)==1 and one[0]['age_days']==7
    edit(c,j,last_contact_date='2026-10-08');assert not items(c,'quote');edit(c,j,last_contact_date='')
    old=one[0]['id'];new=act(c,'quote',q,'revise').json();assert not items(c,'quote');act(c,'quote',new,'share')
    with app.state.store.tenant('oak') as db:db.execute("UPDATE quote_links SET shared_at='2026-10-01T12:00:00+00:00'")
    assert items(c,'quote')[0]['id']!=old
    edit(c,j,status='cancelled');assert not items(c,'quote');edit(c,j,status='new');act(c,'quote',new,'accept');assert not items(c,'quote')

def test_repeat_exact_customer_open_work_recurring_work_and_explicit_dates(app,monkeypatch):
    c=client(app);a=job(c);b=job(c);config(c,repeat_days=30);edit(c,a,status='completed',completed_date='2026-01-01');edit(c,b,status='completed',completed_date='2026-10-27')
    monkeypatch.setattr(workflow,'today',lambda:'2026-10-28');assert [i['job_id'] for i in items(c,'repeat')]==[a['id']]
    body={'service':'Return visit','message':'Invented follow-up','customer_version':next(x['version'] for x in c.get('/api/customers').json() if x['id']==a['customer_id'])}
    # Existing-customer job endpoint binds its exact path identity.
    made=write(c,'/api/customers/'+a['customer_id']+'/jobs',body);assert made.status_code==200,made.text
    assert not items(c,'repeat');edit(c,made.json(),status='cancelled');assert len(items(c,'repeat'))==1
    edit(c,a,completed_date='');assert not items(c,'repeat')
    edit(c,a,status='new');d,_,_=definition(c,app,j=a);edit(c,a,status='completed',completed_date='2026-01-01');assert not items(c,'repeat')
    apply(c,d,{'action':'state','state':'paused','reason':'Pause test repeat work'});assert not items(c,'repeat')

def test_dismissal_survives_private_edits_snooze_reopens_by_date_and_new_work_is_new(app,monkeypatch):
    c=client(app);j=job(c);edit(c,j,next_action='Call before booking',next_action_date='2026-10-01');monkeypatch.setattr(workflow,'today',lambda:'2026-10-28')
    i=items(c,'next_action')[0];i=decide(c,i,'dismiss');edit(c,j,notes='Private edit',priority='high');saved=items(c,'next_action')[0]
    assert saved['id']==i['id'] and saved['effective_state']=='dismissed'
    saved=decide(c,saved,'reopen');saved=decide(c,saved,'snooze',snooze_until='2026-10-29');assert saved['effective_state']=='snoozed'
    monkeypatch.setattr(workflow,'today',lambda:'2026-10-29');assert items(c,'next_action')[0]['effective_state']=='open'
    edit(c,j,next_action='Call about a different task');all_items=items(c,'next_action');assert len(all_items)==2 and sum(i['eligible'] for i in all_items)==1
    new=next(i for i in all_items if i['eligible']);assert new['id']!=saved['id'] and new['version']==0

def test_review_stale_facts_parallel_decisions_idempotency_and_no_delivery(app):
    c=client(app);j=job(c);edit(c,j,next_action='Contact about quote');i=items(c,'next_action')[0];url='/api/opportunities/'+i['id']+'/actions'
    edit(c,j,notes='Another owner changed this job');assert write(c,url,review_body(i)).status_code==409
    i=items(c,'next_action')[0];key=str(uuid.uuid4());body=review_body(i);a=write(c,url,body,key=key);assert a.status_code==200 and a.json()['delivery']=='not_confirmed'
    assert write(c,url,body,key=key).json()==a.json();assert len(c.get('/api/opportunities/'+i['id']).json()['history'])==1
    i=items(c,'next_action')[0];clients=[client(app),client(app)]
    with ThreadPoolExecutor(2) as pool:results=list(pool.map(lambda cc:write(cc,url,review_body(i,'complete')),clients))
    assert sorted(r.status_code for r in results)==[200,409]
    history=c.get('/api/opportunities/'+i['id']).json()['history'];assert [h['action'] for h in history]==['prepare','complete'];assert 'local draft' in history[-1]['snapshot']['draft']
    assert write(c,url,review_body(items(c,'next_action')[0])).status_code==409
    assert c.get('/api/workspace').json()['quotes']==[]
    with app.state.store.tenant('oak') as db:assert db.execute('SELECT COUNT(*) FROM payments').fetchone()[0]==0

def test_feedback_harmless_metadata_preserves_original_permission_and_history(app):
    c=client(app);j,rid,code=prepare(c);respond(c,app,rid,code);publish(c,rid);before=c.get(BASE+rid).json()
    edit(c,j,notes='Changed private notes',priority='urgent',tags=['follow up'],next_action='Internal review',last_contact_date=workflow.today(),postcode_group='AB12',followup_excluded=True)
    after=c.get(BASE+rid).json();assert after==before and c.get(PUBLIC).json()['count']==1
    assert write(TestClient(app,headers=ORIGIN),ACCESS+'access',{'code':code}).status_code==200
    assert not items(c,'review')

@pytest.mark.parametrize('change',[{'status':'in_progress'},{'scheduled_date':'2026-11-15'},{'completed_date':'2026-01-01'},{'archived':True}])
def test_material_job_changes_suspend_without_rewriting_consent_or_answers(app,change):
    c=client(app);j,rid,code=prepare(c);respond(c,app,rid,code);publish(c,rid);before=c.get(BASE+rid).json();edit(c,j,**change);after=c.get(BASE+rid).json()
    assert after['response']==before['response'] and after['history'][:len(before['history'])]==before['history'];assert after['consent']['state']=='suspended' and c.get(PUBLIC).json()['count']==0
    assert write(TestClient(app,headers=ORIGIN),ACCESS+'access',{'code':code}).status_code==401

def test_review_suggestions_follow_exact_feedback_history_without_reinviting(app):
    c=client(app);a=job(c);b=job(c);edit(c,a,status='completed');edit(c,b,status='completed')
    assert {i['customer_id'] for i in items(c,'review')}=={a['customer_id'],b['customer_id']}
    _,rid,code=prepare(c,a);assert [i['job_id'] for i in items(c,'review')]==[b['id']];respond(c,app,rid,code,allow=False)
    action(c,rid,'revoke');assert [i['job_id'] for i in items(c,'review')]==[b['id']];action(c,rid,'archive');assert [i['job_id'] for i in items(c,'review')]==[b['id']]

def test_postcode_group_is_opt_in_explicit_exact_and_stale_sensitive(app):
    c=client(app);a=job(c);visit=job(c);edit(c,a,status='completed',completed_date='2026-01-01');edit(c,visit,status='scheduled',scheduled_date=workflow.today(),postcode_group='AB12')
    assert not items(c,'postcode');config(c,nearby_enabled=True);assert not items(c,'postcode');edit(c,a,postcode_group='AB12');i=items(c,'postcode')[0]
    assert i['related_job_id']==visit['id'] and 'no distance' in i['reason'] and i['postcode_group']=='AB12'
    edit(c,visit,postcode_group='AB13');assert not items(c,'postcode');assert write(c,'/api/opportunities/'+i['id']+'/actions',review_body(i)).status_code==404
    edit(c,visit,postcode_group='AB12');edit(c,a,tags=['no calls']);config(c,excluded_tags=['no calls']);assert not items(c,'postcode') and not items(c,'repeat')

def test_owner_business_staff_portal_and_private_projection_boundaries(app):
    c=client(app);other=client(app,1);j=job(c);edit(c,j,next_action='PRIVATE sentinel next action',tags=['private']);i=items(c,'next_action')[0];decide(c,i)
    staff,_=add_staff(app,c);p,_=portal(app,c,j);guest=TestClient(app,headers=ORIGIN)
    reads=['/api/pipeline','/api/jobs/'+j['id']+'/workflow-history','/api/opportunities','/api/opportunities/'+i['id'],'/api/opportunity-settings','/api/reports/operations?start=2026-10-01&end=2026-10-28']
    for denied in (staff,p,guest):
        for url in reads:assert denied.get(url).status_code in (401,403),url
        assert write(denied,'/api/jobs/'+j['id']+'/workflow',edit_body(c,j),method='put').status_code in (401,403)
        assert write(denied,'/api/opportunities/'+i['id']+'/actions',review_body(i)).status_code in (401,403)
        assert write(denied,'/api/opportunity-settings',c.get('/api/opportunity-settings').json(),method='put').status_code in (401,403)
    assert other.get('/api/pipeline').json()['jobs']==[] and not items(other)
    for url in (reads[1],reads[3]):assert other.get(url).status_code==404
    assert write(other,'/api/jobs/'+j['id']+'/workflow',edit_body(c,j),method='put').status_code==404
    assert write(other,'/api/opportunities/'+i['id']+'/actions',review_body(i)).status_code==404
    for url in ('/api/public/oak-and-vale','/api/public/oak-and-vale/testimonials','/api/portal/oak-and-vale/me'):
        assert 'PRIVATE sentinel' not in p.get(url).text
    config(c,quote_days=22);assert other.get('/api/opportunity-settings').json()['quote_days']==7

def test_uk_report_counts_estimates_invoices_receipts_deposits_and_internal_costs(app):
    c=client(app);j=job(c);q=quote(c,j,10000);inv=invoice(c,q);stages=schedule(c,inv,(3000,7000));version=doc(c,'invoice',inv)['version']
    assert write(c,'/api/invoices/'+inv['id']+'/schedule/'+stages[0]['id']+'/request',{'version':version}).status_code==200
    pay=write(c,'/api/invoices/'+inv['id']+'/payments',{'amount_pence':2500,'paid_on':'2026-10-02','method':'bank'});assert pay.status_code==200,pay.text
    draft=quote(c,job(c),5000);p=person(c);entry(c,j,p,work_date='2026-10-02',duration_minutes=60);material(c,j,status='received',received_date='2026-10-02',quantity='2',cost_price='5.00')
    with app.state.store.tenant('oak') as db:
        db.execute("UPDATE invoices SET issued_at='2026-10-01T23:30:00+00:00'")
        db.execute("UPDATE jobs SET created_at='2026-10-01T23:30:00+00:00'")
    r=c.get('/api/reports/operations?start=2026-10-02&end=2026-10-02');assert r.status_code==200,r.text;v=r.json()
    assert v['period_counts']['jobs_created']==2
    assert v['period_money']=={'issued_invoice_value_pence':10000,'recorded_receipts_pence':2500,'internal_labour_cost_pence':2000,'received_material_cost_pence':1000}
    assert v['current']['open_quote_estimate_pence']==5000 and v['current']['issued_invoice_balance_pence']==7500
    prior=c.get('/api/reports/operations?start=2026-10-01&end=2026-10-01').json();assert prior['period_money']['issued_invoice_value_pence']==0 and prior['current']==v['current']
    correction={'version':doc(c,'invoice',inv)['version'],'reason':'Correct an invented recording error','recording_error':True,'replacement':{'amount_pence':1500,'paid_on':'2026-10-03','method':'bank'}}
    assert write(c,'/api/invoices/'+inv['id']+'/payments/'+pay.json()['id']+'/correct',correction).status_code==200
    assert c.get('/api/reports/operations?start=2026-10-02&end=2026-10-02').json()['period_money']['recorded_receipts_pence']==0
    corrected=c.get('/api/reports/operations?start=2026-10-03&end=2026-10-03').json();assert corrected['period_money']['recorded_receipts_pence']==1500 and corrected['current']['issued_invoice_balance_pence']==8500
    assert 'not recognised revenue' in v['definitions']['issued_invoice_value_pence'] and 'accounting profit' in v['definitions']['internal_costs']
    for period in ('start=2026-02-30&end=2026-03-01','start=2026-10-02&end=2026-10-01','start=2025-01-01&end=2026-10-28'):assert c.get('/api/reports/operations?'+period).status_code==422

def test_unfinished_timer_and_source_reservation_block_archive_without_changes(app):
    from test_fieldwork import captured,draft
    c=client(app);j,people,clock,tid=captured(c,app);q=draft(c,j,kind='quote').json();edit(c,j,status='completed')
    before=doc(c,'quote',q)
    assert write(c,'/api/jobs/'+j['id']+'/workflow',edit_body(c,j,archived=True),method='put').status_code==409
    assert doc(c,'quote',q)==before

def test_expired_shared_quote_and_voided_invoice_do_not_inflate_current_metrics(app,monkeypatch):
    c=client(app);j=job(c);q=quote(c,j);body=edit_payload(doc(c,'quote',q));body['valid_until']=workflow.today();assert write(c,'/api/quotes/'+q['id'],body,method='put').status_code==200;act(c,'quote',q,'share')
    tomorrow=(date.fromisoformat(workflow.today())+timedelta(days=8)).isoformat();monkeypatch.setattr(workflow,'today',lambda:tomorrow);monkeypatch.setattr('backend.operations_report.today',lambda:tomorrow);assert not items(c,'quote')
    j2,q2,i=draft_invoice(c,7000);act(c,'invoice',i,'issue');act(c,'invoice',i,'void')
    r=c.get('/api/reports/operations?start=2026-10-01&end=2026-11-01').json();assert r['current']['open_quote_estimate_pence']==r['current']['issued_invoice_balance_pence']==r['period_money']['issued_invoice_value_pence']==0

def test_settings_and_review_require_fresh_versions_and_deliberate_confirmation(app):
    c=client(app);j=job(c);edit(c,j,next_action='Check invented job');i=items(c,'next_action')[0];url='/api/opportunities/'+i['id']+'/actions'
    for change in ({'confirmed':False},{'confirmed':1},{'draft':'     '},{'action':'dismiss','draft':'','note':'     '}):assert write(c,url,review_body(i,**change)).status_code==422
    assert write(c,url,review_body(i,'reopen')).status_code==409
    old=c.get('/api/opportunity-settings').json();config(c,quote_days=9)
    assert write(c,'/api/opportunity-settings',old,method='put').status_code==409
    assert write(c,url,review_body(i)).status_code==409