"""Fictional recurrence, exact boundaries and deliberately reviewed future visits."""
from concurrent.futures import ThreadPoolExecutor
from datetime import date,datetime
import copy,json,uuid
import pytest
from backend import recurrence_models as m
from backend.recurrence_dates import scheduled
from test_boundaries import app,client,write,job,quote,portal
from test_workforce import person,entry
from test_scheduling import booking,edit_payload
from test_identity import add_staff,member,update

BASE='/api/recurring'
def rule(p,**changes):return {'anchor_date':'2026-10-26','until_date':'2027-12-31','weekdays':[0],'resource_ids':[p['id']],**changes}
def definition(c,app=None,j=None,p=None,**changes):
    if app:app.state.recurrence_today=lambda:date(2026,10,26)
    j=j or job(c);p=p or person(c);version=c.get('/api/jobs/'+j['id']).json()['version']
    r=write(c,BASE,{'name':'Regular care','kind':'job','job_id':j['id'],'customer_id':j['customer_id'],'job_version':version,'effective_from':'2026-10-26','rule':rule(p),**changes});assert r.status_code==200,r.text
    return r.json(),j,p

def current(c,d):return c.get(BASE+'/'+d['id']).json()
def preview(c,d,op):
    d=current(c,d);r=write(c,BASE+'/'+d['id']+'/preview',{'version':d['version'],'operation':op});assert r.status_code==200,r.text;return r.json()
def confirm(c,d,p,**changes):return write(c,BASE+'/'+d['id']+'/confirm',{'version':p['contract_version'],'operation':p['operation'],'preview_hash':p['preview_hash'],'confirmed':True,'acknowledge_retained':True,**changes})
def apply(c,d,op):
    p=preview(c,d,op);assert p['can_confirm'],p;r=confirm(c,d,p);assert r.status_code==200,r.text;return current(c,d)
def generate(c,d,start='2026-11-02',end='2026-11-30'):return apply(c,d,{'action':'generate','start_date':start,'end_date':end})
def single(o,action='cancel',**changes):return {'action':action,'nominal_date':o['nominal_date'],'occurrence_version':o['version'],'booking_version':o['booking']['version'] if o['booking'] else None,'reason':'Owner reviewed future unstarted visit','acknowledge_history':True,**changes}
def revise(d,**changes):return {'action':'revise','effective_from':'2026-11-02','window_end':'2026-11-30','rule':d['rules'][-1]['content'],'name':d['name'],'service_label':d['service_label'],'note':d['note'],'reason':'Review future series',**changes}

def test_definition_only_generate_retry_and_stable_identity(app):
    c=client(app);d,j,p=definition(c,app)
    assert not d['occurrences']
    plan=preview(c,d,{'action':'generate','start_date':'2026-11-02','end_date':'2026-11-16'})
    key=str(uuid.uuid4());args={'version':1,'operation':plan['operation'],'preview_hash':plan['preview_hash'],'confirmed':True}
    a=write(c,BASE+'/'+d['id']+'/confirm',args,key=key);assert a.status_code==200,a.text
    assert write(c,BASE+'/'+d['id']+'/confirm',args,key=key).json()==a.json()
    assert write(c,BASE+'/'+d['id']+'/confirm',args).status_code==409
    d=current(c,d);assert len(d['occurrences'])==3
    assert not preview(c,d,plan['operation'])['can_confirm']
    assert c.get(BASE+'/'+d['id']+'/runs/'+a.json()['run_id']).json()['result']==a.json()
    b=d['occurrences'][0]['booking'];assert write(c,'/api/calendar/'+b['id'],edit_payload(b),method='put').status_code==409
    assert write(c,'/api/calendar/'+b['id']+'/archive',{'version':1}).status_code==409
    listing=c.get('/api/calendar',params={'start_date':'2026-11-01','end_date':'2026-11-30'}).json()['items'];assert all(b['recurrence']['contract_id']==d['id'] for b in listing)


def test_different_concurrent_keys_generate_only_one_set(app):
    c=client(app);d,j,p=definition(c,app);plan=preview(c,d,{'action':'generate','start_date':'2026-11-02','end_date':'2026-11-30'});clients=[client(app),client(app)]
    with ThreadPoolExecutor(2) as pool:responses=list(pool.map(lambda x:confirm(x,d,plan),clients))
    assert sorted(r.status_code for r in responses)==[200,409];assert len(current(c,d)['occurrences'])==5


@pytest.mark.parametrize('change',['calendar','resource','job','customer','date'])
def test_preview_version_fence(app,change):
    c=client(app);d,j,p=definition(c,app);plan=preview(c,d,{'action':'generate','start_date':'2026-11-02','end_date':'2026-11-30'})
    if change=='calendar':booking(c,j,p,start_date='2026-12-01',end_date='2026-12-01')
    elif change=='date':app.state.recurrence_today=lambda:date(2026,10,27)
    else:
        table={'resource':'team_resources','job':'jobs','customer':'customers'}[change];rid={'resource':p['id'],'job':j['id'],'customer':j['customer_id']}[change]
        with app.state.store.tenant('oak') as db:db.execute('UPDATE '+table+' SET version=version+1 WHERE id=?',(rid,))
    assert confirm(c,d,plan).status_code==409;assert not current(c,d)['occurrences']


def test_conflict_rejects_whole_window_without_partial_write(app):
    c=client(app);d,j,p=definition(c,app);booking(c,j,p,start_date='2026-11-09',end_date='2026-11-09',start_time='09:30',end_time='10:30')
    plan=preview(c,d,{'action':'generate','start_date':'2026-11-02','end_date':'2026-11-16'});assert not plan['can_confirm'];assert sum(r['status']=='error' for r in plan['rows'])==1
    assert confirm(c,d,plan).status_code==409;assert not current(c,d)['occurrences']


def test_override_cancel_and_overlapping_exclusions_survive_regeneration(app):
    c=client(app);d,j,p=definition(c,app);d=generate(c,d);original=d['occurrences'][0]
    fields={k:v for k,v in edit_payload(original['booking'],start_time='13:00',end_time='14:00').items() if k!='version'}
    d=apply(c,d,single(original,'override',booking=fields));d=apply(c,d,single(d['occurrences'][1]));cancelled=d['occurrences'][1]
    d=apply(c,d,{'action':'block','start_date':'2026-11-01','end_date':'2026-11-23','reason':'First holiday'})
    assert d['occurrences'][0]['overridden'] and d['occurrences'][0]['state']=='active'
    d=apply(c,d,{'action':'block','start_date':'2026-11-16','end_date':'2026-11-30','reason':'Second holiday'})
    first=d['blocks'][0];d=apply(c,d,{'action':'unblock','block_id':first['id'],'block_version':first['version'],'reason':'First holiday removed'})
    assert d['occurrences'][2]['state']=='excluded';assert d['occurrences'][1]['id']==cancelled['id'] and d['occurrences'][1]['state']=='cancelled'
    second=next(b for b in d['blocks'] if b['active']);d=apply(c,d,{'action':'unblock','block_id':second['id'],'block_version':second['version'],'reason':'Second holiday removed'})
    assert d['occurrences'][2]['state']=='active';assert d['occurrences'][0]['id']==original['id'];assert d['occurrences'][0]['booking']['start_time']=='13:00'
    assert not preview(c,d,{'action':'generate','start_date':'2026-11-02','end_date':'2026-11-30'})['can_confirm']
    d=apply(c,d,single(d['occurrences'][1],'restore'));assert d['occurrences'][1]['booking_id']==cancelled['booking_id']


def test_future_rule_revision_reconciles_outside_new_window_and_keeps_old_rules(app):
    c=client(app);d,j,p=definition(c,app);d=generate(c,d);old=d['rules'][0]
    d=apply(c,d,revise(d,window_end='2026-11-09',rule={**d['rules'][0]['content'],'weekdays':[1]}))
    assert d['rules'][0]==old;assert len(d['rules'])==2
    assert all(o['state']=='excluded' for o in d['occurrences'] if o['nominal_date']>='2026-11-16')
    assert [o['nominal_date'] for o in d['occurrences'] if o['state']=='active']==['2026-11-03']


def test_pause_retire_restore_preserve_visits_and_do_not_auto_generate(app):
    c=client(app);d,j,p=definition(c,app);d=generate(c,d);before=copy.deepcopy(d['occurrences'])
    for state in ('paused','active','retired','restore'):
        d=apply(c,d,{'action':'state','state':state,'reason':'Owner reviewed state'});assert d['occurrences']==before
    assert d['state']=='paused';assert not preview(c,d,{'action':'generate','start_date':'2026-12-01','end_date':'2026-12-31'})['can_confirm']


def test_past_and_ambiguous_today_protected_but_future_visit_can_be_reviewed(app):
    c=client(app);d,j,p=definition(c,app);d=generate(c,d,'2026-10-26','2026-11-16');quote(c,j)
    before=current(c,d)['occurrences'][0];app.state.recurrence_today=lambda:date(2026,11,2)
    d=current(c,d);plan=preview(c,d,revise(d,rule={**d['rules'][0]['content'],'start_time':'12:00'}));assert plan['retained_for_separate_review']==3
    assert confirm(c,d,plan,acknowledge_retained=False).status_code==409
    assert confirm(c,d,plan).status_code==200
    d=current(c,d);assert d['occurrences'][0]==before
    assert not preview(c,d,single(d['occurrences'][0]))['can_confirm']
    assert not preview(c,d,single(d['occurrences'][1]))['can_confirm']
    future=d['occurrences'][2];assert not preview(c,d,single(future,acknowledge_history=False))['can_confirm']
    d=apply(c,d,single(future));assert d['occurrences'][2]['state']=='cancelled'
    d=apply(c,d,{'action':'block','start_date':'2026-11-16','end_date':'2026-11-16','reason':'Holiday on future visit'})
    d=apply(c,d,single(d['occurrences'][3],'exclude'));assert d['occurrences'][3]['state']=='excluded'
    assert not preview(c,d,single(d['occurrences'][3],'restore'))['can_confirm']
    block=d['blocks'][0];d=apply(c,d,{'action':'unblock','block_id':block['id'],'block_version':1,'reason':'Holiday removed'})
    d=apply(c,d,single(d['occurrences'][3],'restore'));assert d['occurrences'][3]['state']=='active'


@pytest.mark.parametrize('frequency,anchor,day,policy,expected',[
('fortnightly','2026-12-28',None,None,['2026-12-28','2027-01-11','2027-01-25']),
('monthly','2027-01-01',31,'skip',['2027-01-31','2027-03-31']),
('monthly','2028-01-01',31,'last_day',['2028-01-31','2028-02-29','2028-03-31']),
('quarterly','2027-01-01',31,'last_day',['2027-01-31','2027-04-30']),
('yearly','2028-02-01',29,'last_day',['2028-02-29','2029-02-28'])])
def test_anchor_month_end_leap_and_intervals(frequency,anchor,day,policy,expected):
    from datetime import timedelta
    value=m.Rule(**rule({'id':'a'*32},frequency=frequency,anchor_date=anchor,until_date='2029-12-31',weekdays=[0] if day is None else [],month_day=day,month_policy=policy));r={'content':value.model_dump()};lo=date.fromisoformat(anchor);hi=date.fromisoformat(expected[-1]);found=[]
    while lo<=hi:
        if scheduled(r,lo.isoformat()):found.append(lo.isoformat())
        lo+=timedelta(days=1)
    assert found==expected


@pytest.mark.parametrize('day,changes,can_confirm,state',[
('2027-03-28',{'start_time':'01:30'},False,None),
('2027-03-28',{'start_time':'01:30','gap_policy':'skip'},True,'excluded'),
('2027-10-31',{'start_time':'01:10','duration_minutes':30},False,None),
('2027-10-31',{'start_time':'01:10','duration_minutes':30,'start_fold':1,'end_fold':1},True,'active'),
('2100-12-31',{'start_time':'23:30','duration_minutes':120},False,None)])
def test_clock_gap_fold_and_upper_bound(app,day,changes,can_confirm,state):
    c=client(app);p=person(c);r=rule(p,anchor_date=day,until_date=day,weekdays=[date.fromisoformat(day).weekday()],**changes);d,j,p=definition(c,app,p=p,rule=r)
    plan=preview(c,d,{'action':'generate','start_date':day,'end_date':day});assert plan['can_confirm']==can_confirm,plan
    if can_confirm:assert confirm(c,d,plan).status_code==200;assert current(c,d)['occurrences'][0]['state']==state


def test_tenant_staff_customer_denials_and_exact_binding(app):
    c=client(app);d,j,p=definition(c,app);other=client(app,1);d=generate(c,d)
    assert other.get(BASE+'/'+d['id']).status_code==404
    staff,_=add_staff(app,c,p['id']);customer,_=portal(app,c,j)
    for who in (staff,customer):
        assert who.get(BASE).status_code in (401,403)
        assert who.get(BASE+'/'+d['id']).status_code in (401,403)
        assert write(who,BASE+'/'+d['id']+'/preview',{'version':d['version'],'operation':{'action':'generate','start_date':'2026-12-01','end_date':'2026-12-31'}}).status_code in (401,403)
    data=staff.get('/api/staff/work').json();assert data['bookings'][0]['is_recurring'];assert 'contract_name' not in json.dumps(data)
    otherjob=job(c);payload={'name':'Wrong customer','job_id':j['id'],'customer_id':otherjob['customer_id'],'job_version':1,'effective_from':'2026-10-26','rule':rule(p)};assert write(c,BASE,payload).status_code==409

@pytest.mark.parametrize('operation',['cancel','exclude','override'])
def test_completed_field_photo_and_document_history_survives_future_change(app,operation):
    from test_fieldwork import captured,base,draft,docact
    from test_photos import reviewed
    from test_recovery import table_rows
    from backend.recurrence import TABLES
    c=client(app);j,people,clock,tid=captured(c,app);p=people[0];photo=reviewed(c,j)
    assert write(c,base(j)+'/logs',{'kind':'note','content':'Completed previous visit','work_date':'2026-10-25','photo_ids':[photo['id']]}).status_code==200
    invoice=draft(c,j).json();assert docact(c,invoice,'issue').status_code==200
    d,j,p=definition(c,app,j,p)
    d=generate(c,d);o=d['occurrences'][0]
    before=table_rows(app.state.store.root,'oak')
    if operation=='exclude':d=apply(c,d,{'action':'block','start_date':o['nominal_date'],'end_date':o['nominal_date'],'reason':'Future holiday reviewed'});o=d['occurrences'][0]
    changes={}
    if operation=='override':changes['booking']={**bookingForTest(o['booking']),'start_date':'2026-11-03','end_date':'2026-11-03','start_time':'13:00','end_time':'14:00'}
    d=apply(c,d,single(o,operation,**changes));after=table_rows(app.state.store.root,'oak')
    for table in ('field_logs','field_log_history','field_sessions','field_intervals','field_history','field_time_links','time_entries','job_photos','photo_history','quotes','invoices','document_sources','source_history'):assert before[table]==after[table],table
    changed=d['occurrences'][0];assert changed['id']==o['id'] and changed['booking_id']==o['booking_id'];assert changed['state']=={'cancel':'cancelled','exclude':'excluded','override':'active'}[operation]


def bookingForTest(b):return {k:v for k,v in edit_payload(b).items() if k!='version'}


def test_last_assignment_loss_interrupts_staff_timer_and_revokes_cached_retry(app):
    from test_fieldwork import setup,start,act,timer,base,review,convert
    c=client(app);j,people,clock=setup(c,app);d,j,p=definition(c,app,j,people[0]);d=generate(c,d,'2026-11-02','2026-11-02');staff,_=add_staff(app,c,p['id'])
    key=str(uuid.uuid4());r=start(staff,j,[p],key=key);assert r.status_code==200;tid=r.json()['id'];clock.advance(60);assert act(staff,j,tid,'heartbeat').status_code==200
    d=apply(c,d,single(d['occurrences'][0]));assert staff.get(base(j)).status_code==404 and start(staff,j,[p],key=key).status_code==404
    t=timer(c,j,tid);assert t['state']=='interrupted' and t['confirmed_ms']==60000
    assert review(c,j,tid).status_code==200 and convert(c,j,tid).status_code==200


def test_other_active_assignment_keeps_staff_access(app):
    from test_fieldwork import setup,start,act,timer
    c=client(app);j,people,clock=setup(c,app);d,j,p=definition(c,app,j,people[0]);d=generate(c,d,'2026-11-02','2026-11-09');staff,_=add_staff(app,c,p['id']);tid=start(staff,j,[p]).json()['id']
    d=apply(c,d,single(d['occurrences'][0]));assert timer(staff,j,tid)['state']=='running';clock.advance(50);assert act(staff,j,tid,'stop').status_code==200


def test_staff_cannot_reuse_an_owner_confirmation_after_demotion(app):
    c=client(app);d,j,p=definition(c,app);owner,_=add_staff(app,c,role='owner');plan=preview(owner,d,{'action':'generate','start_date':'2026-11-02','end_date':'2026-11-02'})
    key=str(uuid.uuid4());args=dict(version=plan['contract_version'],operation=plan['operation'],preview_hash=plan['preview_hash'],confirmed=True)
    assert write(owner,BASE+'/'+d['id']+'/confirm',args,key=key).status_code==200
    assert update(c,member(c),role='staff').status_code==200
    assert write(owner,BASE+'/'+d['id']+'/confirm',args,key=key).status_code in (401,403)


@pytest.mark.parametrize('changes',[{'weekdays':[]},{'weekdays':[0,0]},{'month_day':3},{'until_date':'2033-01-01'},{'frequency':'monthly','weekdays':[]},{'resource_ids':[]},{'duration_minutes':0}])
def test_invalid_or_unbounded_rules_are_rejected(app,changes):
    c=client(app);p=person(c);j=job(c)
    assert write(c,BASE,{'name':'Invalid definition','job_id':j['id'],'customer_id':j['customer_id'],'job_version':1,'effective_from':'2026-10-26','rule':rule(p,**changes)}).status_code==422


def test_whole_team_all_day_and_midnight_exclusion(app):
    c=client(app);p=person(c);r=rule(p,whole_team=True,resource_ids=[],all_day=True,duration_days=1,duration_minutes=None)
    d,j,p=definition(c,app,p=p,kind='unavailable',job_id=None,customer_id=None,job_version=None,rule=r)
    d=generate(c,d,'2026-11-02','2026-11-02');assert d['occurrences'][0]['booking']['whole_team']==1
    other=person(c,name='Later added person');assert write(c,'/api/calendar',{'kind':'job','job_id':j['id'],'resource_ids':[other['id']],'start_date':'2026-11-02','end_date':'2026-11-02','start_time':'15:00','end_time':'16:00'}).status_code==409
    d2,j,p=definition(c,p=other,rule=rule(other,start_time='23:00',duration_minutes=60));d2=generate(c,d2,'2026-11-09','2026-11-09')
    d2=apply(c,d2,{'action':'block','start_date':'2026-11-10','end_date':'2026-11-10','reason':'Next day only'});assert d2['occurrences'][0]['state']=='active'